HCOLL-352 Changed garbage-free Chronicle-Map iterations API from old …

…to new Contexts; Add SegmentContexts API
OpenHFT · Jul 24, 2015 · 9fe1dab · 9fe1dab
1 parent 912a3cc
commit 9fe1dab
Show file tree

Hide file tree

Showing 32 changed files with 196 additions and 231 deletions.
diff --git a/src/main/java/net/openhft/chronicle/hash/ChronicleHash.java b/src/main/java/net/openhft/chronicle/hash/ChronicleHash.java
@@ -33,8 +33,8 @@
  * This interface defines common {@link ChronicleMap} and {@link ChronicleSet}, related to off-heap
  * memory management and file-mapping. Not usable by itself.
  */
-public interface ChronicleHash<K, C extends KeyContext<K>, EQC extends ExternalHashQueryContext<K>>
-        extends Closeable {
+public interface ChronicleHash<K, C extends HashEntry<K>, SC extends HashSegmentContext<K, ?>,
+        EQC extends ExternalHashQueryContext<K>> extends Closeable {
     /**
      * Returns the file this hash container mapped to, i. e. when it is created by
      * {@link ChronicleHashBuilder#create()} call, or {@code null} if it is purely in-memory,
@@ -81,6 +81,8 @@ public interface ChronicleHash<K, C extends KeyContext<K>, EQC extends ExternalH
      * @return the context to perform operations with the key
      */
     @NotNull EQC queryContext(Data<K> key);
+
+    SC segmentContext(int segmentIndex);
 
     /**
      * Checks the given predicate on each entry in this {@code ChronicleHash} until all entries

diff --git a/src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilder.java b/src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilder.java
@@ -65,7 +65,7 @@
  * @param <B> the concrete builder type, i. e. {@link ChronicleMapBuilder}
  *            or {@link ChronicleSetBuilder}
  */
-public interface ChronicleHashBuilder<K, H extends ChronicleHash<K, ?, ?>,
+public interface ChronicleHashBuilder<K, H extends ChronicleHash<K, ?, ?, ?>,
         B extends ChronicleHashBuilder<K, H, B>> extends Cloneable {
 
     /**

diff --git a/src/main/java/net/openhft/chronicle/hash/HashContext.java b/src/main/java/net/openhft/chronicle/hash/HashContext.java
@@ -27,5 +27,5 @@ public interface HashContext<K> {
     /**
      * Returns the accessed {@code ChronicleHash}.
      */
-    ChronicleHash<K, ?, ?> hash();
+    ChronicleHash<K, ?, ?, ?> hash();
 }
diff --git a/src/main/java/net/openhft/chronicle/hash/HashSegmentContext.java b/src/main/java/net/openhft/chronicle/hash/HashSegmentContext.java
@@ -0,0 +1,33 @@
+/*
+ * Copyright 2015 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.hash;
+
+import net.openhft.chronicle.core.io.Closeable;
+
+import java.util.function.Consumer;
+import java.util.function.Predicate;
+
+public interface HashSegmentContext<K, E extends HashEntry<K>> extends HashContext<K>, Closeable {
+
+    void forEachSegmentEntry(Consumer<? super E> action);
+
+    boolean forEachSegmentEntryWhile(Predicate<? super E> predicate);
+
+    long size();
+}
diff --git a/src/main/java/net/openhft/chronicle/hash/impl/ChronicleHashBuilderImpl.java b/src/main/java/net/openhft/chronicle/hash/impl/ChronicleHashBuilderImpl.java
@@ -22,7 +22,7 @@
 import net.openhft.chronicle.hash.serialization.internal.SerializationBuilder;
 
 //TODO remove this temporary interface
-public interface ChronicleHashBuilderImpl<K, H extends ChronicleHash<K, ?, ?>,
+public interface ChronicleHashBuilderImpl<K, H extends ChronicleHash<K, ?, ?, ?>,
         B extends ChronicleHashBuilder<K, H, B>> extends ChronicleHashBuilder<K, H, B> {
 
     SerializationBuilder<K> keyBuilder();

diff --git a/src/main/java/net/openhft/chronicle/hash/impl/VanillaChronicleHash.java b/src/main/java/net/openhft/chronicle/hash/impl/VanillaChronicleHash.java
@@ -16,9 +16,7 @@
 
 package net.openhft.chronicle.hash.impl;
 
-import net.openhft.chronicle.hash.ChronicleHash;
-import net.openhft.chronicle.hash.ExternalHashQueryContext;
-import net.openhft.chronicle.hash.KeyContext;
+import net.openhft.chronicle.hash.*;
 import net.openhft.chronicle.hash.impl.hashlookup.HashLookup;
 import net.openhft.chronicle.hash.impl.util.BuildVersion;
 import net.openhft.chronicle.hash.serialization.BytesReader;
@@ -44,8 +42,9 @@
 import static net.openhft.lang.MemoryUnit.*;
 
 public abstract class VanillaChronicleHash<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
-        C extends KeyContext<K>, ECQ extends ExternalHashQueryContext<K>>
-        implements ChronicleHash<K, C, ECQ>, Serializable {
+        C extends HashEntry<K>, SC extends HashSegmentContext<K, ?>,
+        ECQ extends ExternalHashQueryContext<K>>
+        implements ChronicleHash<K, C, SC, ECQ>, Serializable {
 
     private static final long serialVersionUID = 0L;
 

diff --git a/src/main/java/net/openhft/chronicle/hash/impl/VanillaChronicleHashHolder.java b/src/main/java/net/openhft/chronicle/hash/impl/VanillaChronicleHashHolder.java
@@ -21,5 +21,5 @@
 import net.openhft.chronicle.hash.serialization.internal.MetaBytesInterop;
 
 public interface VanillaChronicleHashHolder<K, KI, MKI extends MetaBytesInterop<K, ? super KI>> {
-    VanillaChronicleHash<K, KI, MKI, ?, ?> h();
+    VanillaChronicleHash<K, KI, MKI, ?, ?, ?> h();
 }
diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/SegmentStages.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/SegmentStages.java
@@ -360,7 +360,7 @@ boolean segmentInit() {
     }
 
     void initSegment() {
-        VanillaChronicleHash<?, ?, ?, ?, ?> h = hh.h();
+        VanillaChronicleHash<?, ?, ?, ?, ?, ?> h = hh.h();
         long hashLookupOffset = h.segmentOffset(segmentIndex);
         long freeListOffset = hashLookupOffset + h.segmentHashLookupOuterSize;
         freeListBytes.storePositionAndSize(h.ms, freeListOffset, h.segmentFreeListInnerSize);
@@ -375,7 +375,7 @@ void closeSegment() {
 
     //TODO refactor/optimize
     public long alloc(int chunks) {
-        VanillaChronicleHash<?, ?, ?, ?, ?> h = hh.h();
+        VanillaChronicleHash<?, ?, ?, ?, ?, ?> h = hh.h();
         if (chunks > h.maxChunksPerEntry)
             throw new IllegalArgumentException("Entry is too large: requires " + chunks +
                     " entry size chucks, " + h.maxChunksPerEntry + " is maximum.");

diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/iter/HashSegmentIteration.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/iter/HashSegmentIteration.java
@@ -27,6 +27,7 @@
 import net.openhft.sg.StageRef;
 import net.openhft.sg.Staged;
 
+import java.util.function.Consumer;
 import java.util.function.Predicate;
 
 @Staged
@@ -50,7 +51,7 @@ protected void initEntryRemovedOnThisIteration(boolean entryRemovedOnThisIterati
         this.entryRemovedOnThisIteration = entryRemovedOnThisIteration;
     }
 
-    public boolean forEachRemoving(Predicate<? super E> action) {
+    public boolean forEachSegmentEntryWhile(Predicate<? super E> action) {
         s.innerUpdateLock.lock();
         try {
             long size = s.size();
@@ -86,6 +87,13 @@ public boolean forEachRemoving(Predicate<? super E> action) {
             initEntryRemovedOnThisIteration(false);
         }
     }
+
+    public void forEachSegmentEntry(Consumer<? super E> action) {
+        forEachSegmentEntryWhile(e -> {
+            action.accept(e);
+            return true;
+        });
+    }
 
     public void checkEntryNotRemovedOnThisIteration() {
         if (entryRemovedOnThisIterationInit())

diff --git a/src/main/java/net/openhft/chronicle/map/AbstractChronicleMap.java b/src/main/java/net/openhft/chronicle/map/AbstractChronicleMap.java
@@ -80,7 +80,7 @@ default void putAll(File fromFile) throws IOException {
 
     @Override
     default boolean containsValue(Object value) {
-        return !forEachEntryWhile(c -> !c.valueEqualTo((V) value));
+        return !forEachEntryWhile(c -> !c.value().equals(c.context().wrapValueAsData((V) value)));
     }
 
     @Override
@@ -90,7 +90,7 @@ default boolean isEmpty() {
 
     @Override
     default void forEach(BiConsumer<? super K, ? super V> action) {
-        forEachEntry(c -> action.accept(c.key(), c.get()));
+        forEachEntry(c -> action.accept(c.key().get(), c.value().get()));
     }
 
     @Override
@@ -147,7 +147,7 @@ public boolean contains(Object v) {
 
             @Override
             public void forEach(java.util.function.Consumer<? super V> action) {
-                AbstractChronicleMap.this.forEachEntry(c -> action.accept(c.get()));
+                AbstractChronicleMap.this.forEachEntry(c -> action.accept(c.value().get()));
             }
         };
     }
@@ -181,7 +181,7 @@ public boolean contains(Object k) {
 
             @Override
             public void forEach(java.util.function.Consumer<? super K> action) {
-                AbstractChronicleMap.this.forEachEntry(c -> action.accept(c.key()));
+                AbstractChronicleMap.this.forEachEntry(c -> action.accept(c.key().get()));
             }
         };
     }
@@ -198,9 +198,9 @@ default boolean mapEquals(Object o) {
 
         try {
             return forEachEntryWhile(c -> {
-                K k = c.key();
+                K k = c.key().get();
                 V v = (V) m.get(k instanceof CharSequence ? k.toString() : k);
-                return v != null && c.valueEqualTo(v);
+                return v != null && c.value().equals(c.context().wrapValueAsData(v));
             });
         } catch (ClassCastException unused) {
             return false;
@@ -247,25 +247,21 @@ default Set<Entry<K, V>> newEntrySet() {
 
 
     @Override
-    default void forEachEntry(final Consumer<? super MapKeyContext<K, V>> action) {
+    default void forEachEntry(final Consumer<? super MapEntry<K, V>> action) {
         forEachEntryWhile(c -> {
             action.accept(c);
             return true;
         });
     }
-
-    IterationContextInterface<K, V> iterationContext();
 
     @Override
-    default boolean forEachEntryWhile(final Predicate<? super MapKeyContext<K, V>> action) {
+    default boolean forEachEntryWhile(final Predicate<? super MapEntry<K, V>> action) {
         boolean interrupt = false;
-        iteration:
-        try (IterationContextInterface<K, V> c = iterationContext()) {
-            for (int segmentIndex = actualSegments() - 1; segmentIndex >= 0; segmentIndex--) {
-                c.initTheSegmentIndex(segmentIndex);
-                if (!c.forEachRemoving(e -> action.test(c.deprecatedMapKeyContextOnIteration()))) {
+        for (int i = actualSegments() - 1; i >= 0; i--) {
+            try (MapSegmentContext<K, V, ?> c = segmentContext(i)) {
+                if (!c.forEachSegmentEntryWhile(action)) {
                     interrupt = true;
-                    break iteration;
+                    break;
                 }
             }
         }

diff --git a/src/main/java/net/openhft/chronicle/map/ChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ChronicleMap.java
@@ -30,7 +30,6 @@
 import java.io.File;
 import java.io.IOException;
 import java.io.Serializable;
-import java.lang.Object;
 import java.util.Map;
 import java.util.concurrent.ConcurrentMap;
 
@@ -54,9 +53,9 @@
  * @param <K> the map key type
  * @param <V> the map value type
  */
-public interface ChronicleMap<K, V>
-        extends ConcurrentMap<K, V>,
-        ChronicleHash<K, MapKeyContext<K, V>, ExternalMapQueryContext<K, V, ?>> {
+public interface ChronicleMap<K, V> extends ConcurrentMap<K, V>,
+        ChronicleHash<K, MapEntry<K, V>, MapSegmentContext<K, V, ?>,
+                ExternalMapQueryContext<K, V, ?>> {
 
     /**
      * Returns the value to which the specified key is mapped, or {@code null} if this map contains

diff --git a/src/main/java/net/openhft/chronicle/map/ChronicleMapIterator.java b/src/main/java/net/openhft/chronicle/map/ChronicleMapIterator.java
@@ -16,24 +16,22 @@
 
 package net.openhft.chronicle.map;
 
-import net.openhft.chronicle.map.impl.IterationContextInterface;
-
 import java.util.ArrayDeque;
 import java.util.Iterator;
 import java.util.Map.Entry;
 import java.util.NoSuchElementException;
 import java.util.Queue;
+import java.util.function.Consumer;
 
 /**
  * Very inefficient (esp. if segments are large), but CORRECT implementation
  */
-abstract class ChronicleMapIterator<K, V, E> implements Iterator<E> {
+abstract class ChronicleMapIterator<K, V, E> implements Iterator<E>, Consumer<MapEntry<K, V>> {
 
     final AbstractChronicleMap<K, V> map;
     private final Thread ownerThread = Thread.currentThread();
     private final Queue<E> entryBuffer = new ArrayDeque<>();
     private int segmentIndex;
-    IterationContextInterface<K, V> context;
     E returned;
 
     ChronicleMapIterator(AbstractChronicleMap<K, V> map) {
@@ -54,24 +52,23 @@ private void fillEntryBuffer() {
         while (true) {
             if (segmentIndex < 0)
                 return;
-            try (IterationContextInterface<K, V> c = map.iterationContext()) {
-                context = c;
-                c.initTheSegmentIndex(segmentIndex);
+
+            try (MapSegmentContext<K, V, ?> c = map.segmentContext(segmentIndex)) {
                 segmentIndex--;
                 if (c.size() == 0)
                     continue;
-                c.forEachRemoving(e -> {
-                    entryBuffer.add(read());
-                    return true;
-                });
+                c.forEachSegmentEntry(this);
                 return;
-            } finally {
-                context = null;
             }
         }
     }
 
-    abstract E read();
+    @Override
+    public void accept(MapEntry<K, V> e) {
+        entryBuffer.add(read(e));
+    }
+
+    abstract E read(MapEntry<K, V> entry);
 
     @Override
     public boolean hasNext() {
@@ -108,9 +105,9 @@ static class OfEntries<K, V> extends ChronicleMapIterator<K, V, Entry<K, V>> {
         }
 
         @Override
-        Entry<K, V> read() {
-            K key = context.key().getUsing(null);
-            V value = context.value().getUsing(null);
+        Entry<K, V> read(MapEntry<K, V> entry) {
+            K key = entry.key().getUsing(null);
+            V value = entry.value().getUsing(null);
             return new WriteThroughEntry<>(map, key, value);
         }
 
@@ -127,8 +124,8 @@ static class OfKeys<K, V> extends ChronicleMapIterator<K, V, K> {
         }
 
         @Override
-        K read() {
-            return context.key().getUsing(null);
+        K read(MapEntry<K, V> entry) {
+            return entry.key().getUsing(null);
         }
 
         @Override

diff --git a/src/main/java/net/openhft/chronicle/map/MapSegmentContext.java b/src/main/java/net/openhft/chronicle/map/MapSegmentContext.java
@@ -0,0 +1,25 @@
+/*
+ * Copyright 2015 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.chronicle.hash.HashSegmentContext;
+
+public interface MapSegmentContext<K, V, R>
+        extends HashSegmentContext<K, MapEntry<K, V>>, MapContext<K, V, R> {
+}
diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java
@@ -18,7 +18,6 @@
 
 package net.openhft.chronicle.map;
 
-import net.openhft.chronicle.hash.KeyContext;
 import net.openhft.chronicle.hash.replication.AbstractReplication;
 import net.openhft.chronicle.hash.replication.ReplicableEntry;
 import net.openhft.chronicle.hash.replication.TimeProvider;
@@ -218,7 +217,7 @@ public void onHeaderCreated() {
 
     @Override
     public void clear() {
-        forEachEntry(KeyContext::remove);
+        forEachEntry(c -> c.context().remove(c));
     }
 
     void addCloseable(Closeable closeable) {