diff --git a/src/main/java/net/openhft/chronicle/hash/serialization/internal/BytesReaders.java b/src/main/java/net/openhft/chronicle/hash/serialization/internal/BytesReaders.java index 477f09f1d..8fe37b984 100644 --- a/src/main/java/net/openhft/chronicle/hash/serialization/internal/BytesReaders.java +++ b/src/main/java/net/openhft/chronicle/hash/serialization/internal/BytesReaders.java @@ -34,32 +34,33 @@ public final class BytesReaders { * from {@link net.openhft.chronicle.hash.serialization} package. * * @param marshaller the actual reading implementation - * @param type of the objects marshalled + * @param type of the objects marshalled * @return a {@code BytesReader} wrapping the given {@code BytesMarshaller} */ - public static BytesReader fromBytesMarshaller(BytesMarshaller marshaller) { + public static BytesReader fromBytesMarshaller(BytesMarshaller marshaller) { return new SimpleBytesReader(marshaller); } private static class SimpleBytesReader implements BytesReader { private static final long serialVersionUID = 0L; - private final BytesMarshaller marshaller; + private final BytesMarshaller marshaller; - public SimpleBytesReader(BytesMarshaller marshaller) { + public SimpleBytesReader(BytesMarshaller marshaller) { this.marshaller = marshaller; } @Override public E read(Bytes bytes, long size) { - return marshaller.read(bytes); + return (E) marshaller.read(bytes); } @Override public E read(Bytes bytes, long size, E toReuse) { - return marshaller.read(bytes, toReuse); + return (E) marshaller.read(bytes, toReuse); } } - private BytesReaders() {} + private BytesReaders() { + } } diff --git a/src/main/java/net/openhft/chronicle/map/AbstractChronicleMapBuilder.java b/src/main/java/net/openhft/chronicle/map/AbstractChronicleMapBuilder.java index a9eb45115..915337413 100755 --- a/src/main/java/net/openhft/chronicle/map/AbstractChronicleMapBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/AbstractChronicleMapBuilder.java @@ -39,7 +39,6 @@ import net.openhft.lang.io.serialization.impl.VanillaBytesMarshallerFactory; import net.openhft.lang.model.Byteable; import net.openhft.lang.threadlocal.Provider; import net.openhft.lang.threadlocal.ThreadLocalCopies; -import net.openhft.lang.values.LongValue; import org.jetbrains.annotations.NotNull; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -985,7 +984,7 @@ abstract class AbstractChronicleMapBuilder> 10), + return (int) Math.min(Math.max(4L, entries() >> 10), VanillaChronicleMap.MAX_ENTRY_OVERSIZE_FACTOR); } diff --git a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java index f19bb87e6..017702283 100755 --- a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java @@ -271,7 +271,7 @@ public final class ChronicleMapBuilder implements ChronicleMapBuilderI valueMarshaller( - @NotNull BytesMarshaller valueMarshaller) { + @NotNull BytesMarshaller valueMarshaller) { delegate.valueMarshaller(valueMarshaller); return this; } diff --git a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilderI.java b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilderI.java index ed4dd05d5..8268a2008 100755 --- a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilderI.java +++ b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilderI.java @@ -199,7 +199,7 @@ public interface ChronicleMapBuilderI extends Serializable { * @see #objectSerializer(ObjectSerializer) */ ChronicleMapBuilderI valueMarshaller( - @NotNull BytesMarshaller valueMarshaller); + @NotNull BytesMarshaller valueMarshaller); /** * Configures the marshallers, used to serialize/deserialize values to/from off-heap memory in diff --git a/src/main/java/net/openhft/chronicle/map/OffHeapUpdatableChronicleMapBuilder.java b/src/main/java/net/openhft/chronicle/map/OffHeapUpdatableChronicleMapBuilder.java index 8dc117593..e94a9383a 100644 --- a/src/main/java/net/openhft/chronicle/map/OffHeapUpdatableChronicleMapBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/OffHeapUpdatableChronicleMapBuilder.java @@ -148,7 +148,7 @@ final class OffHeapUpdatableChronicleMapBuilder return super.prepareValueBytesOnAcquire(prepareValueBytes); } - public ChronicleMapBuilderI valueMarshaller(@NotNull BytesMarshaller valueMarshaller) { + public ChronicleMapBuilderI valueMarshaller(@NotNull BytesMarshaller valueMarshaller) { throw new UnsupportedOperationException("not supported for this combination of key/value type"); } diff --git a/src/main/java/net/openhft/chronicle/map/OnHeapUpdatableChronicleMapBuilder.java b/src/main/java/net/openhft/chronicle/map/OnHeapUpdatableChronicleMapBuilder.java index eaed43fdf..058dc94cb 100644 --- a/src/main/java/net/openhft/chronicle/map/OnHeapUpdatableChronicleMapBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/OnHeapUpdatableChronicleMapBuilder.java @@ -210,7 +210,7 @@ final class OnHeapUpdatableChronicleMapBuilder * @see #objectSerializer(ObjectSerializer) */ public OnHeapUpdatableChronicleMapBuilder valueMarshaller( - @NotNull BytesMarshaller valueMarshaller) { + @NotNull BytesMarshaller valueMarshaller) { valueBuilder.marshaller(valueMarshaller); return this; } diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java index eebe5cc7a..60a3357ef 100644 --- a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java @@ -231,7 +231,7 @@ final class ReplicatedChronicleMap, RV, VB extends RV, VBI, MVBI extends MetaBytesInterop> - RV put(ThreadLocalCopies copies, SegmentState segmentState, - MKBI metaKeyInterop, KBI keyInterop, KB key, long keySize, - InstanceOrBytesToInstance toKey, - GetValueInterops getValueInterops, VB value, - InstanceOrBytesToInstance toValue, - boolean replaceIfPresent, ReadValue readValue, boolean resultUnused) { + RV put2(ThreadLocalCopies copies, SegmentState segmentState, + MKBI metaKeyInterop, KBI keyInterop, KB key, long keySize, + InstanceOrBytesToInstance toKey, + GetValueInterops getValueInterops, VB value, + InstanceOrBytesToInstance toValue, + boolean replaceIfPresent, ReadValue readValue, boolean resultUnused) { return put(copies, segmentState, metaKeyInterop, keyInterop, key, keySize, toKey, getValueInterops, value, toValue, replaceIfPresent, readValue, resultUnused, localIdentifier, currentTime()); diff --git a/src/main/java/net/openhft/chronicle/map/SerializationBuilder.java b/src/main/java/net/openhft/chronicle/map/SerializationBuilder.java index 34818fe75..7c87d2ee9 100644 --- a/src/main/java/net/openhft/chronicle/map/SerializationBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/SerializationBuilder.java @@ -18,9 +18,6 @@ package net.openhft.chronicle.map; -import net.openhft.lang.model.*; -import net.openhft.lang.threadlocal.Provider; -import net.openhft.lang.threadlocal.ThreadLocalCopies; import net.openhft.chronicle.hash.serialization.*; import net.openhft.chronicle.hash.serialization.internal.*; import net.openhft.lang.io.Bytes; @@ -29,6 +26,12 @@ import net.openhft.lang.io.serialization.BytesMarshaller; import net.openhft.lang.io.serialization.ObjectFactory; import net.openhft.lang.io.serialization.ObjectSerializer; import net.openhft.lang.io.serialization.impl.*; +import net.openhft.lang.model.Byteable; +import net.openhft.lang.model.DataValueGenerator; +import net.openhft.lang.model.DataValueModel; +import net.openhft.lang.model.DataValueModels; +import net.openhft.lang.threadlocal.Provider; +import net.openhft.lang.threadlocal.ThreadLocalCopies; import net.openhft.lang.values.LongValue; import org.jetbrains.annotations.NotNull; import org.jetbrains.annotations.Nullable; @@ -44,11 +47,11 @@ import java.util.Arrays; import java.util.List; import static net.openhft.chronicle.hash.serialization.SizeMarshallers.constant; -import static net.openhft.chronicle.map.Objects.hash; import static net.openhft.chronicle.hash.serialization.SizeMarshallers.stopBit; +import static net.openhft.chronicle.map.Objects.hash; import static net.openhft.chronicle.map.SerializationBuilder.Role.KEY; -final class SerializationBuilder implements Cloneable, Serializable{ +final class SerializationBuilder implements Cloneable, Serializable { private static boolean concreteClass(Class c) { return !c.isInterface() && !Modifier.isAbstract(c.getModifiers()); @@ -194,7 +197,7 @@ final class SerializationBuilder implements Cloneable, Serializable{ .>forBytesWriter(role)); } - public SerializationBuilder marshaller(BytesMarshaller marshaller) { + public SerializationBuilder marshaller(BytesMarshaller marshaller) { clearDefaults(); return copyingInterop(CopyingInterop.FROM_MARSHALLER) .reader(BytesReaders.fromBytesMarshaller(marshaller)) diff --git a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java index d83b83753..439c381ca 100755 --- a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java @@ -354,7 +354,7 @@ class VanillaChronicleMap, void putBytes(ThreadLocalCopies copies, SegmentState segmentState, Bytes key, long keySize, GetRemoteBytesValueInterops getRemoteBytesValueInterops, MultiStoreBytes value, boolean replaceIfPresent, ReadValue readValue) { - put(copies, segmentState, + put2(copies, segmentState, DelegatingMetaBytesInterop.>instance(), BytesBytesInterop.INSTANCE, key, keySize, keyBytesToInstance, getRemoteBytesValueInterops, value, valueBytesToInstance, @@ -373,7 +373,7 @@ class VanillaChronicleMap, getRemoteBytesValueInterops.valueSizeMarshaller(valueSizeMarshaller); ReadValueToBytes readValueToLazyBytes = segmentState.readValueToLazyBytes; readValueToLazyBytes.valueSizeMarshaller(valueSizeMarshaller); - put(copies, segmentState, + put2(copies, segmentState, DelegatingMetaBytesInterop.>instance(), BytesBytesInterop.INSTANCE, entry, keySize, keyBytesToInstance, getRemoteBytesValueInterops, value, valueBytesToInstance, @@ -453,15 +453,15 @@ class VanillaChronicleMap, @Override public final V put(K key, V value) { - return put0(key, value, true); + return put1(key, value, true); } @Override public final V putIfAbsent(@net.openhft.lang.model.constraints.NotNull K key, V value) { - return put0(key, value, false); + return put1(key, value, false); } - V put0(@net.openhft.lang.model.constraints.NotNull K key, V value, boolean replaceIfPresent) { + V put1(@net.openhft.lang.model.constraints.NotNull K key, V value, boolean replaceIfPresent) { checkKey(key); checkValue(value); ThreadLocalCopies copies = keyInteropProvider.getCopies(null); @@ -470,22 +470,22 @@ class VanillaChronicleMap, MKI metaKeyInterop = metaKeyInteropProvider.get(copies, originalMetaKeyInterop, keyInterop, key); long keySize = metaKeyInterop.size(keyInterop, key); - return put(copies, null, metaKeyInterop, keyInterop, key, keySize, keyIdentity(), + return put2(copies, null, metaKeyInterop, keyInterop, key, keySize, keyIdentity(), this, value, valueIdentity(), replaceIfPresent, this, putReturnsNull); } , RV, VB extends RV, VBI, MVBI extends MetaBytesInterop> - RV put(ThreadLocalCopies copies, SegmentState segmentState, - MKBI metaKeyInterop, KBI keyInterop, KB key, long keySize, - InstanceOrBytesToInstance toKey, - GetValueInterops getValueInterops, VB value, - InstanceOrBytesToInstance toValue, - boolean replaceIfPresent, ReadValue readValue, boolean resultUnused) { + RV put2(ThreadLocalCopies copies, SegmentState segmentState, + MKBI metaKeyInterop, KBI keyInterop, KB key, long keySize, + InstanceOrBytesToInstance toKey, + GetValueInterops getValueInterops, VB value, + InstanceOrBytesToInstance toValue, + boolean replaceIfPresent, ReadValue readValue, boolean resultUnused) { long hash = metaKeyInterop.hash(keyInterop, key); int segmentNum = getSegment(hash); long segmentHash = segmentHash(hash); - return segments[segmentNum].put(copies, segmentState, + return segments[segmentNum].put3(copies, segmentState, metaKeyInterop, keyInterop, key, keySize, toKey, getValueInterops, value, toValue, segmentHash, replaceIfPresent, readValue, resultUnused); @@ -1729,15 +1729,15 @@ class VanillaChronicleMap, return v; } - private , + , RV, VB extends RV, VBI, MVBI extends MetaBytesInterop> - RV put(@Nullable ThreadLocalCopies copies, final @Nullable SegmentState segmentState, - MKBI metaKeyInterop, KBI keyInterop, KB key, long keySize, - InstanceOrBytesToInstance toKey, - GetValueInterops getValueInterops, VB value, - InstanceOrBytesToInstance toValue, - long hash2, boolean replaceIfPresent, - ReadValue readValue, boolean resultUnused) { + RV put3(@Nullable ThreadLocalCopies copies, final @Nullable SegmentState segmentState, + MKBI metaKeyInterop, KBI keyInterop, KB key, long keySize, + InstanceOrBytesToInstance toKey, + GetValueInterops getValueInterops, VB value, + InstanceOrBytesToInstance toValue, + long hash2, boolean replaceIfPresent, + ReadValue readValue, boolean resultUnused) { shouldNotBeCalledFromReplicatedChronicleMap("Segment.put"); writeLock(null); try { diff --git a/src/test/java/eg/BGChronicleTest.java b/src/test/java/eg/BGChronicleTest.java index 1b65540ef..436d93ab2 100644 --- a/src/test/java/eg/BGChronicleTest.java +++ b/src/test/java/eg/BGChronicleTest.java @@ -22,10 +22,10 @@ public class BGChronicleTest { public static final int CHRONICLE_HEARTBEAT_SECONDS = 5; public static final int READ_RATIO = Integer.getInteger("readRatio", 2); - // public static final boolean BUSY_WAIT = Boolean.parseBoolean(System.getProperty("busyWait", -// "" + (CLIENTS < Runtime.getRuntime().availableProcessors()))); public static final int REPLICAS = Integer.getInteger("replicas", 1); public static final int CLIENTS = Integer.getInteger("clients", Math.max(1, Runtime.getRuntime().availableProcessors() - REPLICAS)); + public static final boolean BUSY_WAIT = Boolean.parseBoolean(System.getProperty("busyWaiting", + "" + (CLIENTS + REPLICAS < Runtime.getRuntime().availableProcessors()))); public static final long KEYS = Long.getLong("keys", 1000_000L); public static final long MESSAGES = Long.getLong("messages", 1000_000 + KEYS * 2); public static final long MAX_RATE = Long.getLong("maxRate", 100000); @@ -232,8 +232,8 @@ public class BGChronicleTest { if (pause > 0) Thread.sleep(pause); // wait for residual. -// if (BUSY_WAIT) -// while (System.nanoTime() < next) ; + if (BUSY_WAIT) + while (System.nanoTime() < next) ; // 100 possible values for the first byte. long start; key[0] = b; diff --git a/src/test/java/net/openhft/chronicle/map/LargeEntriesTest.java b/src/test/java/net/openhft/chronicle/map/LargeEntriesTest.java new file mode 100644 index 000000000..12b73c8c2 --- /dev/null +++ b/src/test/java/net/openhft/chronicle/map/LargeEntriesTest.java @@ -0,0 +1,147 @@ +/* + * Copyright 2014 Higher Frequency Trading + * + * http://www.higherfrequencytrading.com + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package net.openhft.chronicle.map; + +import net.openhft.lang.io.DirectBytes; +import net.openhft.lang.io.DirectStore; +import net.openhft.lang.io.serialization.impl.SnappyStringMarshaller; +import org.junit.Ignore; +import org.junit.Test; + +import java.beans.XMLEncoder; +import java.io.ByteArrayOutputStream; +import java.util.*; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; + +/** + * Created by peter on 06/12/14. + */ +public class LargeEntriesTest { + + static final int ENTRIES = 640; + static final int ENTRY_SIZE = 1024; + + @Test + @Ignore + public void testLargeStrings() throws ExecutionException, InterruptedException { + final ChronicleMap map = ChronicleMapBuilder + .of(String.class, String.class) + .valueMarshaller(SnappyStringMarshaller.INSTANCE) + .actualSegments(1) // to force an error. + .entries(ENTRIES * 2) + .entrySize(ENTRY_SIZE / 4) + .putReturnsNull(true) + .create(); + { + int threads = 2; //Runtime.getRuntime().availableProcessors(); + ExecutorService es = Executors.newFixedThreadPool(threads); + final int block = ENTRIES / threads; + for (int i = 0; i < 3; i++) { + long start = System.currentTimeMillis(); + List> futureList = new ArrayList<>(); + for (int t = 0; t < threads; t++) { + final int finalT = t; + final int finalI = i; + futureList.add(es.submit(new Runnable() { + @Override + public void run() { + exerciseLargeStrings(map, finalI, finalT * block, finalT * block + block, ENTRY_SIZE); + } + })); + } + for (Future future : futureList) { + future.get(); + } + long time = System.currentTimeMillis() - start; + long operations = 3; + System.out.printf("Put/Get rate was %.1f MB/s%n", operations * ENTRIES * ENTRY_SIZE / 1e6 / (time / 1e3)); + } + es.shutdown(); + if (es.isTerminated()) + map.close(); + } + } + + void exerciseLargeStrings(ChronicleMap map, int run, int start, int finish, int entrySize) { +/* + final Thread thisThread = Thread.currentThread(); + Thread monitor = new Thread(new Runnable() { + @Override + public void run() { + StringBuilder sb = new StringBuilder(); + sb.append(thisThread).append("\n"); + try { + while (!Thread.currentThread().isInterrupted()) { + Thread.sleep(50); + StackTraceElement[] stackTrace = thisThread.getStackTrace(); + for (int i = 0; i < 4 && i < stackTrace.length; i++) + sb.append("\tat ").append(stackTrace[i]).append("\n"); + sb.append("\n"); + } + } catch (InterruptedException e) { + } + System.out.println(sb); + } + }); + monitor.start(); +*/ + ByteArrayOutputStream baos = new ByteArrayOutputStream(); + XMLEncoder xml = new XMLEncoder(baos); + Map map2 = new HashMap<>(); + Random rand = new Random(1); + for (int i = 0; i < entrySize / 80; i++) + map2.put("key-" + i, "value-" + rand.nextInt(1000)); + xml.writeObject(map2); + xml.close(); + String value = baos.toString().substring(0, entrySize); + // warmup to compression. + if (run == 0) { + DirectBytes bytes = DirectStore.allocate(entrySize).bytes(); + SnappyStringMarshaller.INSTANCE.write(bytes, value); + bytes.flip(); + SnappyStringMarshaller.INSTANCE.read(bytes); + bytes.release(); + } + + for (int i = start; i < finish; i++) { + String key = "key-" + i; + map.put(key, value); + String object = map.get(key); + + assertNotNull(key, object); + assertEquals(key, entrySize, object.length()); + } +// monitor.interrupt(); + + for (int i = start; i < finish; i++) { +// System.out.println(i); + String key = "key-" + i; + String object = map.get(key); + + assertNotNull(key, object); + assertEquals(key, entrySize, object.length()); + } + } +}