Merge remote-tracking branch 'origin/master'

src/main/java/net/openhft/chronicle/hash/replication/TcpTransportAndNetworkConfig.java

@@ -178,7 +178,8 @@ public final class TcpTransportAndNetworkConfig implements Serializable {
         if (endpoints != null ? !endpoints.equals(that.endpoints) : that.endpoints != null)
             return false;
         if (heartBeatIntervalUnit != that.heartBeatIntervalUnit) return false;
-        if (throttlingConfig != null ? !throttlingConfig.equals(that.throttlingConfig) : that.throttlingConfig != null)
+        if (throttlingConfig != null ? !throttlingConfig.equals(that.throttlingConfig) :
+                that.throttlingConfig != null)
             return false;
 
         return true;

src/main/java/net/openhft/chronicle/hash/serialization/internal/CopyingMetaBytesInterop.java

@@ -20,6 +20,7 @@ package net.openhft.chronicle.hash.serialization.internal;
 
 import net.openhft.chronicle.hash.serialization.BytesWriter;
 import net.openhft.chronicle.hash.hashing.Hasher;
+import net.openhft.lang.MemoryUnit;
 import net.openhft.lang.io.Bytes;
 import net.openhft.lang.io.DirectBytes;
 import net.openhft.lang.io.DirectStore;
@@ -36,6 +37,19 @@ import java.io.Serializable;
 public abstract class CopyingMetaBytesInterop<E, W> implements MetaBytesInterop<E, W> {
     private static final long serialVersionUID = 0L;
 
+    private static final long MAX_REASONABLE_SERIALIZED_SIZE = MemoryUnit.MEGABYTES.toBytes(512L);
+
+    public static void checkMaxSizeStillReasonable(long maxSize, Exception ex) {
+        if (maxSize > MAX_REASONABLE_SERIALIZED_SIZE) {
+            throw new IllegalStateException("We try to figure out size of objects " +
+                    "in serialized form, but it exceeds " +
+                    MAX_REASONABLE_SERIALIZED_SIZE + " bytes. We assume this is " +
+                    "a error and throw exception at this point. If you really " +
+                    "want larger keys/values, use ChronicleMapBuilder." +
+                    "keySize(int)/valueSize(int)/entrySize(int) configurations", ex);
+        }
+    }
+
     final DirectBytesBuffer buffer;
     transient long size;
     transient long hash;
@@ -50,11 +64,19 @@ public abstract class CopyingMetaBytesInterop<E, W> implements MetaBytesInterop<
         if (mutable || writer != this.writer || e != cur) {
             this.writer = writer;
             cur = e;
-            Bytes buffer = this.buffer.obtain(maxSize);
-            innerWrite(writer, buffer, e);
-            buffer.flip();
-            size = buffer.remaining();
-            hash = 0L;
+            while (true) {
+                try {
+                    Bytes buffer = this.buffer.obtain(maxSize);
+                    innerWrite(writer, buffer, e);
+                    buffer.flip();
+                    size = buffer.remaining();
+                    hash = 0L;
+                    return;
+                } catch (Exception ex) {
+                    checkMaxSizeStillReasonable(maxSize, ex);
+                    maxSize *= 2L;
+                }
+            }
         }
     }
 

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

@@ -51,6 +51,7 @@ import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import static net.openhft.chronicle.map.Objects.builderEquals;
+import static net.openhft.chronicle.map.VanillaChronicleMap.MAX_ENTRY_OVERSIZE_FACTOR;
 import static net.openhft.lang.model.DataValueGenerator.firstPrimitiveFieldType;
 
 /**
@@ -407,17 +408,32 @@ public final class ChronicleMapBuilder<K, V> implements Cloneable,
 
     @Override
     public ChronicleMapBuilder<K, V> maxEntryOversizeFactor(int maxEntryOversizeFactor) {
-        if (maxEntryOversizeFactor < 1 || maxEntryOversizeFactor > 64)
-            throw new IllegalArgumentException("maxEntryOversizeFactor should be in [1, 64] " +
-                    "range, " + maxEntryOversizeFactor + " given");
+        if (maxEntryOversizeFactor < 1)
+            throw new IllegalArgumentException("maxEntryOversizeFactor should be more than 1, " +
+                    maxEntryOversizeFactor + " given");
         this.maxEntryOversizeFactor = maxEntryOversizeFactor;
         return this;
     }
 
-    int maxEntryOversizeFactor() {
-        if (maxEntryOversizeFactor < 0)
-            return keyBuilder.constantSizeMarshaller() && valueBuilder.constantSizeMarshaller() ?
-                    1 : 64;
+    private void checkMaxEntryOversizeFactorIfReplicated(boolean replicated) {
+        if (replicated && maxEntryOversizeFactor > MAX_ENTRY_OVERSIZE_FACTOR)
+            throw new IllegalStateException("As of now Replicated Chornicle Map doesn't support " +
+                    "more than " + MAX_ENTRY_OVERSIZE_FACTOR + " entry oversize factor, " +
+                    maxEntryOversizeFactor + " is configured");
+    }
+
+    int maxEntryOversizeFactor(boolean replicated) {
+        if (maxEntryOversizeFactor < 0) {
+            if (keyBuilder.constantSizeMarshaller() && valueBuilder.constantSizeMarshaller()) {
+                return 1;
+            } else {
+                if (replicated) {
+                    return MAX_ENTRY_OVERSIZE_FACTOR;
+                } else {
+                    return Integer.MAX_VALUE;
+                }
+            }
+        }
         return maxEntryOversizeFactor;
     }
 
@@ -1233,7 +1249,7 @@ public final class ChronicleMapBuilder<K, V> implements Cloneable,
         keyBuilder.objectSerializer(acquireObjectSerializer(JDKObjectSerializer.INSTANCE));
         valueBuilder.objectSerializer(acquireObjectSerializer(JDKObjectSerializer.INSTANCE));
 
-        long maxSize = (long) entrySize(replicated) * figureBufferAllocationFactor();
+        long maxSize = (long) entrySize(replicated) * figureBufferAllocationFactor(replicated);
         keyBuilder.maxSize(maxSize);
         valueBuilder.maxSize(maxSize);
 
@@ -1241,12 +1257,13 @@ public final class ChronicleMapBuilder<K, V> implements Cloneable,
             keyBuilder.constantSizeBySample(sampleKey);
         if (sampleValue != null)
             valueBuilder.constantSizeBySample(sampleValue);
-        stateChecks();
+        stateChecks(replicated);
     }
 
-    private void stateChecks() {
+    private void stateChecks(boolean replicated) {
         checkAlignmentOnlyIfValuesPossiblyReferenceOffHeap();
         checkPrepareValueBytesOnlyIfConstantValueSize();
+        checkMaxEntryOversizeFactorIfReplicated(replicated);
     }
 
     private ChronicleMap<K, V> establishReplication(
@@ -1288,11 +1305,11 @@ public final class ChronicleMapBuilder<K, V> implements Cloneable,
         return map;
     }
 
-    private int figureBufferAllocationFactor() {
+    private int figureBufferAllocationFactor(boolean replicated) {
         // if expected map size is about 1000, seems rather wasteful to allocate
         // key and value serialization buffers each x64 of expected entry size..
         return (int) Math.min(Math.max(2L, entries() >> 10),
-                VanillaChronicleMap.MAX_ENTRY_OVERSIZE_FACTOR);
+                maxEntryOversizeFactor(replicated));
     }
 }
 

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

@@ -427,13 +427,7 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
         long start = entry.position();
         final long keySize = keySizeMarshaller.readSize(entry);
 
-        entry.skip(keySize);
-
-        // timeStamp
-        entry.readLong();
-
-        //identifier
-        entry.readByte();
+        entry.skip(keySize + ADDITIONAL_ENTRY_BYTES - 1L);
 
         final boolean isDeleted = entry.readBoolean();
         long valueSize;
@@ -444,8 +438,7 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
         }
 
         alignment.alignPositionAddr(entry);
-        entry.skip(valueSize);
-        long result = (entry.position() - start);
+        long result = (entry.position() + valueSize - start);
         entry.position(start);
 
         // entries can be larger than Integer.MAX_VALUE as we are restricted to the size we can

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

@@ -21,6 +21,7 @@ package net.openhft.chronicle.map;
 import net.openhft.chronicle.hash.serialization.*;
 import net.openhft.chronicle.hash.serialization.internal.*;
 import net.openhft.chronicle.hash.serialization.internal.ByteBufferMarshaller;
+import net.openhft.lang.MemoryUnit;
 import net.openhft.lang.io.ByteBufferBytes;
 import net.openhft.lang.io.Bytes;
 import net.openhft.lang.io.serialization.BytesMarshallable;
@@ -261,10 +262,7 @@ final class SerializationBuilder<E> implements Cloneable, Serializable {
                             copies, this.metaInterop, interop, sampleObject);
                     break findSufficientSerializationSize;
                 } catch (Exception e) {
-                    // assuming nobody need > 512 mb _constant size_ keys/values
-                    if (maxSize > 512 * 1024 * 1024) {
-                        throw e;
-                    }
+                    CopyingMetaBytesInterop.checkMaxSizeStillReasonable(maxSize, e);
                     maxSize(maxSize * 2);
                 }
             }

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

@@ -35,7 +35,7 @@ class StatelessMapConfig<K, V> implements StatelessClientConfig<ChronicleMap<K,
     private final long timeoutMs;
     private final String name;
 
-    private int tcpBufferSize = 1024 * 64;
+    private int tcpBufferSize = 1024 * VanillaChronicleMap.MAX_ENTRY_OVERSIZE_FACTOR;
 
     private final AtomicBoolean used;
 

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

@@ -177,7 +177,8 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
         this.eventListener = builder.eventListener();
         this.bytesEventListener = builder.bytesEventListener();
         this.segmentHeaderSize = builder.segmentHeaderSize();
-        this.maxEntryOversizeFactor = builder.maxEntryOversizeFactor();
+        this.maxEntryOversizeFactor =
+                builder.maxEntryOversizeFactor(this instanceof ReplicatedChronicleMap);
 
         hashSplitting = HashSplitting.Splitting.forSegments(actualSegments);
         initTransients();

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

@@ -0,0 +1,36 @@
+/*
+ * 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 org.junit.Assert;
+import org.junit.Test;
+
+public class NoEntryOversizeFactorTest {
+
+    @Test
+    public void noOversizeFactorTest() {
+        ChronicleMap<Integer, CharSequence> map =
+                ChronicleMapBuilder.of(Integer.class, CharSequence.class)
+                        .valueSize(1).entries(10000L).actualSegments(1).create();
+        String ultraLargeValue = "";
+        for (int i = 0; i < 100; i++) {
+            ultraLargeValue += "Hello";
+        }
+        map.put(1, ultraLargeValue);
+        Assert.assertEquals(ultraLargeValue, map.get(1));
+    }
+}