Merge remote-tracking branch 'origin/master'

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

@@ -33,8 +33,7 @@ public interface DefaultValueProvider<K, V> extends Serializable {
      * for the specified key, if it is absent in the map.
      *
      * @param key key absent in the map
-     * @return value to be put for the specified key in the map, or {@code null}, if the key should
-     * remain absent in the map
+     * @return value to be put for the specified key in the map
      */
     V get(K key);
 }

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

@@ -91,7 +91,7 @@ public final class OffHeapUpdatableChronicleMapBuilder<K, V>
      * @see #defaultValueProvider(DefaultValueProvider)
      */
     @Override
-    OffHeapUpdatableChronicleMapBuilder<K, V> prepareValueBytesOnAcquire(
+    public OffHeapUpdatableChronicleMapBuilder<K, V> prepareValueBytesOnAcquire(
             @NotNull PrepareValueBytes<K> prepareValueBytes) {
         return super.prepareValueBytesOnAcquire(prepareValueBytes);
     }

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

@@ -25,6 +25,7 @@ import net.openhft.chronicle.common.serialization.MetaBytesInterop;
 import net.openhft.chronicle.common.serialization.MetaBytesWriter;
 import net.openhft.chronicle.common.threadlocal.ThreadLocalCopies;
 import net.openhft.lang.Maths;
+import net.openhft.lang.MemoryUnit;
 import net.openhft.lang.collection.ATSDirectBitSet;
 import net.openhft.lang.io.Bytes;
 import net.openhft.lang.io.MultiStoreBytes;
@@ -41,6 +42,9 @@ import java.util.concurrent.CopyOnWriteArraySet;
 import java.util.concurrent.atomic.AtomicReferenceArray;
 
 import static net.openhft.chronicle.common.serialization.Hasher.hash;
+import static net.openhft.lang.MemoryUnit.BITS;
+import static net.openhft.lang.MemoryUnit.BYTES;
+import static net.openhft.lang.MemoryUnit.CACHE_LINES;
 import static net.openhft.lang.collection.DirectBitSet.NOT_FOUND;
 
 /**
@@ -120,7 +124,7 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
     }
 
     private int assignedModIterBitSetSizeInBytes() {
-        return (int) align64((127 + RESERVED_MOD_ITER) / 8);
+        return (int) CACHE_LINES.align(BYTES.alignAndConvert(127 + RESERVED_MOD_ITER, BITS), BYTES);
     }
 
     @Override
@@ -139,7 +143,8 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
     }
 
     long modIterBitSetSizeInBytes() {
-        return align64(bitsPerSegmentInModIterBitSet() * segments.length / 8L);
+        long bytes = BITS.toBytes(bitsPerSegmentInModIterBitSet() * segments.length);
+        return CACHE_LINES.align(bytes, BYTES);
     }
 
     private long bitsPerSegmentInModIterBitSet() {
@@ -587,7 +592,7 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
         @Override
         void createHashLookups(long start) {
             hashLookupLiveAndDeleted = createMultiMap(start);
-            start += align64(sizeOfMultiMap() + sizeOfMultiMapBitSet());
+            start += CACHE_LINES.align(sizeOfMultiMap() + sizeOfMultiMapBitSet(), BYTES);
             hashLookupLiveOnly = createMultiMap(start);
         }
 
@@ -1533,79 +1538,4 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
         }
     }
 
-}
-
-abstract class Replicator {
-
-    /**
-     * A constructor for use in subclasses.
-     */
-    protected Replicator() {
-    }
-
-    /**
-     * Applies the replicator to the map instance and returns a Closeable token to manage resources,
-     * associated with the replication.  <p>This method isn't intended to be called from the
-     * client code.
-     *
-     * @param builder             the builder from which the map was constructed. The replicator may
-     *                            obtain some map configurations, not accessible via the map
-     *                            instance.
-     * @param map                 a replicated map instance. Provides basic tools for replication
-     *                            implementation.
-     * @param entryExternalizable the callback for ser/deser implementation in the replicator
-     * @param chronicleMap        to wrap.
-     * @return a {@code Closeable} token to control replication resources. It should be closed on
-     * closing the replicated map.
-     * @throws java.io.IOException   if an io error occurred during the replicator setup
-     * @throws IllegalStateException if this replicator doesn't support application to more than one
-     *                               map (or the specified number of maps), and this replicator has
-     *                               already been applied to a map (or the specified number of
-     *                               maps)
-     */
-    protected abstract Closeable applyTo(AbstractChronicleMapBuilder builder,
-                                         Replica map, Replica.EntryExternalizable entryExternalizable,
-                                         final ChronicleMap chronicleMap)
-            throws IOException;
-}
-
-
-final class Replicators {
-
-    private Replicators() {
-    }
-
-    static Replicator tcp(final TcpReplicationConfig replicationConfig) {
-        return new Replicator() {
-
-            @Override
-            protected Closeable applyTo(AbstractChronicleMapBuilder builder,
-                                        Replica replica, Replica.EntryExternalizable entryExternalizable,
-                                        final ChronicleMap chronicleMap) throws IOException {
-
-                final KeyValueSerializer keyValueSerializer = new KeyValueSerializer(builder
-                        .keyBuilder, builder.valueBuilder);
-
-                StatelessServerConnector statelessServer = new StatelessServerConnector
-                        (keyValueSerializer, chronicleMap, builder.entrySize());
-
-                return new TcpReplicator(replica, entryExternalizable, replicationConfig,
-                        builder.entrySize(), statelessServer);
-            }
-        };
-    }
-
-    static Replicator udp(
-            final UdpReplicationConfig replicationConfig) {
-        return new Replicator() {
-
-            @Override
-            protected Closeable applyTo(AbstractChronicleMapBuilder builder,
-                                        Replica map, Replica.EntryExternalizable entryExternalizable, final ChronicleMap chronicleMap)
-                    throws IOException {
-                return new UdpReplicator(map, entryExternalizable, replicationConfig,
-                        builder.entrySize());
-            }
-        };
-    }
-}
+}

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

@@ -0,0 +1,54 @@
+/*
+ * 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 java.io.Closeable;
+import java.io.IOException;
+
+abstract class Replicator {
+
+    /**
+     * A constructor for use in subclasses.
+     */
+    protected Replicator() {
+    }
+
+    /**
+     * Applies the replicator to the map instance and returns a Closeable token to manage resources,
+     * associated with the replication.  <p>This method isn't intended to be called from the
+     * client code.
+     *
+     * @param builder             the builder from which the map was constructed. The replicator may
+     *                            obtain some map configurations, not accessible via the map
+     *                            instance.
+     * @param map                 a replicated map instance. Provides basic tools for replication
+     *                            implementation.
+     * @param entryExternalizable the callback for ser/deser implementation in the replicator
+     * @param chronicleMap        to wrap.
+     * @return a {@code Closeable} token to control replication resources. It should be closed on
+     * closing the replicated map.
+     * @throws java.io.IOException   if an io error occurred during the replicator setup
+     * @throws IllegalStateException if this replicator doesn't support application to more than one
+     *                               map (or the specified number of maps), and this replicator has
+     *                               already been applied to a map (or the specified number of
+     *                               maps)
+     */
+    protected abstract Closeable applyTo(AbstractChronicleMapBuilder builder,
+                                         Replica map, Replica.EntryExternalizable entryExternalizable,
+                                         final ChronicleMap chronicleMap)
+            throws IOException;
+}

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

@@ -0,0 +1,60 @@
+/*
+ * 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 java.io.Closeable;
+import java.io.IOException;
+
+final class Replicators {
+
+    private Replicators() {
+    }
+
+    static Replicator tcp(final TcpReplicationConfig replicationConfig) {
+        return new Replicator() {
+
+            @Override
+            protected Closeable applyTo(AbstractChronicleMapBuilder builder,
+                                        Replica replica, Replica.EntryExternalizable entryExternalizable,
+                                        final ChronicleMap chronicleMap) throws IOException {
+
+                final KeyValueSerializer keyValueSerializer = new KeyValueSerializer(builder
+                        .keyBuilder, builder.valueBuilder);
+
+                StatelessServerConnector statelessServer = new StatelessServerConnector
+                        (keyValueSerializer, chronicleMap, builder.entrySize());
+
+                return new TcpReplicator(replica, entryExternalizable, replicationConfig,
+                        builder.entrySize(), statelessServer);
+            }
+        };
+    }
+
+    static Replicator udp(
+            final UdpReplicationConfig replicationConfig) {
+        return new Replicator() {
+
+            @Override
+            protected Closeable applyTo(AbstractChronicleMapBuilder builder,
+                                        Replica map, Replica.EntryExternalizable entryExternalizable, final ChronicleMap chronicleMap)
+                    throws IOException {
+                return new UdpReplicator(map, entryExternalizable, replicationConfig,
+                        builder.entrySize());
+            }
+        };
+    }
+}

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

@@ -570,14 +570,6 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
 
     }
 
-    @NotNull
-    public synchronized Set<K> keySet() {
-        long sizeLocation = writeEvent(KEY_SET);
-
-        // get the data back from the server
-        return readKeySet(blockingFetch(sizeLocation));
-    }
-
 
     @NotNull
     public synchronized Collection<V> values() {
@@ -637,6 +629,42 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
     }
 
 
+    @NotNull
+    public synchronized Set<K> keySet() {
+        final long sizeLocation = writeEvent(KEY_SET);
+
+        final long startTime = System.currentTimeMillis();
+        final long transactionId = nextUniqueTransaction(startTime);
+        final long timeoutTime = System.currentTimeMillis() + builder.timeoutMs();
+
+        // get the data back from the server
+        Bytes in = blockingFetch0(sizeLocation, transactionId, startTime);
+
+        final Set<K> result = new HashSet<>();
+
+        for (; ; ) {
+
+            boolean hasMoreEntries = in.readBoolean();
+            LOG.info("entrySet - hasMoreEntries=" + hasMoreEntries);
+
+            // number of entries in the chunk
+            long size = in.readInt();
+
+            for (int i = 0; i < size; i++) {
+                K k = keyValueSerializer.readKey(in);
+                result.add(k);
+            }
+
+            if (!hasMoreEntries)
+                break;
+
+            in = blockingFetchReadOnly(timeoutTime, transactionId);
+        }
+
+        return result;
+    }
+
+
     private long writeEvent(EventId event) {
 
         buffer.clear();

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

@@ -548,7 +548,7 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable {
             if (completed)
                 attached.entryWriter.workCompleted();
 
-              attached.hasRemoteHeartbeatInterval = false;
+            attached.hasRemoteHeartbeatInterval = false;
 
         } else if (attached.remoteModificationIterator != null)
             attached.entryWriter.entriesToBuffer(attached.remoteModificationIterator, key);
@@ -1573,40 +1573,6 @@ class StatelessServerConnector<K, V> {
         return null;
     }
 
-    private Work keySet(Bytes reader, final Bytes writer) {
-        final long sizeLocation = reflectTransactionId(reader, writer);
-        final Set<K> ks;
-        try {
-            ks = map.keySet();
-        } catch (RuntimeException e) {
-            return sendException(writer, sizeLocation, e);
-        }
-
-        writer.writeStopBit(ks.size());
-
-        final Iterator<K> iterator = ks.iterator();
-
-        // this allows us to write more data than the buffer will allow
-        return new Work() {
-
-            @Override
-            public boolean doWork(@NotNull final Bytes writer) {
-
-                while (iterator.hasNext()) {
-
-                    // we've filled up the buffer, so lets give another channel a chance to send
-                    // some data, we don't know the max key size, we will use the entrySize instead
-                    if (writer.remaining() <= maxEntrySizeBytes)
-                        return false;
-
-                    writeKey(iterator.next(), writer);
-                }
-
-                writeSizeAndFlags(sizeLocation, false, writer);
-                return true;
-            }
-        };
-    }
 
     private Work values(Bytes reader, Bytes writer) {
         final long sizeLocation = reflectTransactionId(reader, writer);
@@ -1644,6 +1610,50 @@ class StatelessServerConnector<K, V> {
         };
     }
 
+    private Work keySet(Bytes reader, final Bytes writer) {
+        final long transactionId = reader.readLong();
+
+        Set<K> ks;
+
+
+        try {
+            ks = map.keySet();
+        } catch (RuntimeException e) {
+            return sendException(reader, writer, e);
+        }
+
+
+        final Iterator<K> iterator = ks.iterator();
+
+        // this allows us to write more data than the buffer will allow
+        return new Work() {
+
+            @Override
+            public boolean doWork(Bytes writer) {
+
+                final long sizeLocation = header(writer, transactionId);
+
+                int count = 0;
+                while (iterator.hasNext()) {
+
+                    // we've filled up the buffer, so lets give another channel a chance to send
+                    // some data, we don't know the max key size, we will use the entrySize instead
+                    if (writer.remaining() <= maxEntrySizeBytes) {
+                        writeHeader(writer, sizeLocation, count, true);
+                        return false;
+                    }
+
+                    count++;
+                    K key = iterator.next();
+                    writeKey(key, writer);
+                }
+
+                writeHeader(writer, sizeLocation, count, false);
+                return true;
+            }
+        };
+    }
+
     private Work entrySet(Bytes reader, Bytes writer) {
 
         final long transactionId = reader.readLong();

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

@@ -22,6 +22,7 @@ import net.openhft.chronicle.common.ChronicleHashErrorListener;
 import net.openhft.chronicle.common.serialization.*;
 import net.openhft.chronicle.common.threadlocal.Provider;
 import net.openhft.chronicle.common.threadlocal.ThreadLocalCopies;
+import net.openhft.lang.MemoryUnit;
 import net.openhft.lang.collection.DirectBitSet;
 import net.openhft.lang.collection.SingleThreadedDirectBitSet;
 import net.openhft.lang.io.*;
@@ -42,7 +43,10 @@ import java.nio.channels.FileChannel;
 import java.util.*;
 import java.util.concurrent.TimeUnit;
 
+import static java.lang.Long.numberOfTrailingZeros;
+import static java.lang.Math.max;
 import static java.lang.Thread.currentThread;
+import static net.openhft.lang.MemoryUnit.*;
 
 
 class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
@@ -90,8 +94,6 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
     final boolean putReturnsNull;
     final boolean removeReturnsNull;
 
-    final boolean useSmallMultiMaps;
-
     private final long lockTimeOutNS;
     private final ChronicleHashErrorListener errorListener;
     transient Segment[] segments; // non-final for close()
@@ -142,7 +144,6 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
         this.eventListener = builder.eventListener();
 
         hashSplitting = HashSplitting.Splitting.forSegments(actualSegments);
-        useSmallMultiMaps = useSmallMultiMaps();
         initTransients();
     }
 
@@ -154,17 +155,6 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
         return hashSplitting.segmentIndex(hash);
     }
 
-    /**
-     * Cache line alignment, assuming 64-byte cache lines.
-     */
-    static long align64(long l) {
-        return (l + 63L) & ~63L;
-    }
-
-    static long align8(long n) {
-        return (n + 7L) & ~7L;
-    }
-
     void initTransients() {
         keyReaderProvider = Provider.of((Class) originalKeyReader.getClass());
         keyInteropProvider = Provider.of((Class) originalKeyInterop.getClass());
@@ -240,23 +230,23 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
     }
 
     long sizeOfMultiMap() {
-        return useSmallMultiMaps ?
+        return useSmallMultiMaps() ?
                 VanillaShortShortMultiMap.sizeInBytes(entriesPerSegment) :
                 VanillaIntIntMultiMap.sizeInBytes(entriesPerSegment);
     }
 
     long sizeOfMultiMapBitSet() {
-        return useSmallMultiMaps ?
+        return useSmallMultiMaps() ?
                 VanillaShortShortMultiMap.sizeOfBitSetInBytes(entriesPerSegment) :
                 VanillaIntIntMultiMap.sizeOfBitSetInBytes(entriesPerSegment);
     }
 
     boolean useSmallMultiMaps() {
-        return entriesPerSegment * 2L <= VanillaShortShortMultiMap.MAX_CAPACITY;
+        return entriesPerSegment <= VanillaShortShortMultiMap.MAX_CAPACITY;
     }
 
     long sizeOfBitSets() {
-        return align64(align8(entriesPerSegment) / 8L);
+        return CACHE_LINES.align(BYTES.alignAndConvert(entriesPerSegment, BITS), BYTES);
     }
 
     int numberOfBitSets() {
@@ -267,28 +257,25 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
 
     long segmentSize() {
         long ss = SEGMENT_HEADER
-                + align64(sizeOfMultiMap() + sizeOfMultiMapBitSet()) * multiMapsPerSegment()
+                + (CACHE_LINES.align(sizeOfMultiMap() + sizeOfMultiMapBitSet(), BYTES)
+                    * multiMapsPerSegment())
                 + numberOfBitSets() * sizeOfBitSets() // the free list and 0+ dirty lists.
                 + sizeOfEntriesInSegment();
         if ((ss & 63L) != 0)
             throw new AssertionError();
+        return breakL1CacheAssociativityContention(ss);
+    }
 
-        // Say, there is 32 KB L1 cache with 2(4, 8) way set associativity, 64-byte lines.
-        // It means there are 32 * 1024 / 64 / 2(4, 8) = 256(128, 64) sets,
-        // i. e. each way (bank) contains 256(128, 64) lines. (L2 and L3 caches has more sets.)
-        // If segment size in lines multiplied by 2^n is divisible by set size,
-        // every 2^n-th segment header fall into the same set.
-        // To break this up we make segment size odd in lines, in this case only each
-        // 256(128, 64)-th segment header fall into the same set.
-
-        // If there are 64 sets in L1, it should be 8- or much less likely 4-way, and segments
-        // collision by pairs is not so terrible.
-
-        // if the size is a multiple of 4096 or slightly more. Make sure it is at least 64 more than a multiple.
-        if ((ss & 4093) < 64)
-            ss = (ss & ~63) + 64;
-
-        return ss;
+    private long breakL1CacheAssociativityContention(long segmentSize) {
+        // Conventional alignment to break is 4096 (given Intel's 32KB 8-way L1 cache),
+        // for any case break 2 times smaller alignment
+        int alignmentToBreak = 2048;
+        int eachNthSegmentFallIntoTheSameSet =
+                max(1, alignmentToBreak >> numberOfTrailingZeros(segmentSize));
+        if (eachNthSegmentFallIntoTheSameSet < actualSegments) {
+            segmentSize |= 64L; // make segment size "odd" (in cache lines)
+        }
+        return segmentSize;
     }
 
     int multiMapsPerSegment() {
@@ -296,7 +283,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
     }
 
     private long sizeOfEntriesInSegment() {
-        return align64(entriesPerSegment * entrySize);
+        return CACHE_LINES.align(entriesPerSegment * entrySize, BYTES);
     }
 
     @Override
@@ -564,9 +551,12 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
 
             long start = bytes.startAddr() + SEGMENT_HEADER;
             createHashLookups(start);
-            start += align64(sizeOfMultiMap() + sizeOfMultiMapBitSet()) * multiMapsPerSegment();
+            start += CACHE_LINES.align(sizeOfMultiMap() + sizeOfMultiMapBitSet(), BYTES)
+                        * multiMapsPerSegment();
             final NativeBytes bsBytes = new NativeBytes(ms.objectSerializer(),
-                    start, start + align8(align8(entriesPerSegment) / 8L), null);
+                    start,
+                    start + LONGS.align(BYTES.alignAndConvert(entriesPerSegment, BITS), BYTES),
+                    null);
             freeList = new SingleThreadedDirectBitSet(bsBytes);
             start += numberOfBitSets() * sizeOfBitSets();
             entriesOffset = start - bytes.startAddr();
@@ -591,7 +581,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
                     new NativeBytes(new VanillaBytesMarshallerFactory(), start,
                             start + sizeOfMultiMapBitSet(), null);
             multiMapBytes.load();
-            return useSmallMultiMaps ?
+            return useSmallMultiMaps() ?
                     new VanillaShortShortMultiMap(multiMapBytes, sizeOfMultiMapBitSetBytes) :
                     new VanillaIntIntMultiMap(multiMapBytes, sizeOfMultiMapBitSetBytes);
         }
@@ -628,7 +618,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
          */
         long getSize() {
             // any negative value is in error state.
-            return Math.max(0L, this.bytes.readVolatileLong(SIZE_OFFSET));
+            return max(0L, this.bytes.readVolatileLong(SIZE_OFFSET));
         }