diff --git a/pom.xml b/pom.xml
index 47741d9d9..3bf9688c2 100644
--- a/pom.xml
+++ b/pom.xml
@@ -197,7 +197,7 @@
http://openhft.github.io/Java-Lang/apidocs/
- net.openhft.chronicle.hash.serialization.internal:net.openhft.xstream:net.openhft.xstream.*
+ net.openhft.chronicle.hash.serialization.internal:net.openhft.xstream:net.openhft.xstream.*:net.openhft.chronicle.hash.impl.*
diff --git a/src/main/java/net/openhft/chronicle/hash/ChronicleHash.java b/src/main/java/net/openhft/chronicle/hash/ChronicleHash.java
index 60f9e9a17..8dcd0766f 100644
--- a/src/main/java/net/openhft/chronicle/hash/ChronicleHash.java
+++ b/src/main/java/net/openhft/chronicle/hash/ChronicleHash.java
@@ -18,13 +18,13 @@
package net.openhft.chronicle.hash;
-import net.openhft.chronicle.hash.function.Consumer;
-import net.openhft.chronicle.hash.function.Predicate;
import net.openhft.chronicle.map.ChronicleMap;
import net.openhft.chronicle.set.ChronicleSet;
import java.io.Closeable;
import java.io.File;
+import java.util.function.Consumer;
+import java.util.function.Predicate;
/**
* This interface defines common {@link ChronicleMap} and {@link ChronicleSet}, related to off-heap
diff --git a/src/main/java/net/openhft/chronicle/hash/function/Function.java b/src/main/java/net/openhft/chronicle/hash/function/Function.java
deleted file mode 100644
index bc7e3724d..000000000
--- a/src/main/java/net/openhft/chronicle/hash/function/Function.java
+++ /dev/null
@@ -1,26 +0,0 @@
-package net.openhft.chronicle.hash.function;
-
-import java.io.Serializable;
-
-/**
- * Represents a function that accepts one argument and produces a result.
- *
- *
This is a functional interface whose functional method is {@link #apply(Object)}.
- *
- *
This is a copy of Java 8's {@code java.util.function.Function} interface.
- *
- * @param the type of the input to the function
- * @param the type of the result of the function
- */
-
-public interface Function extends Serializable {
-
- /**
- * Applies this function to the given argument.
- *
- * @param t the function argument
- * @return the function result
- */
- R apply(T t);
-
-}
diff --git a/src/main/java/net/openhft/chronicle/hash/function/SerializableFunction.java b/src/main/java/net/openhft/chronicle/hash/function/SerializableFunction.java
new file mode 100644
index 000000000..f830b24d6
--- /dev/null
+++ b/src/main/java/net/openhft/chronicle/hash/function/SerializableFunction.java
@@ -0,0 +1,10 @@
+package net.openhft.chronicle.hash.function;
+
+import java.io.Serializable;
+import java.util.function.Function;
+
+/**
+ * {@link Function} + {@link Serializable}
+ */
+public interface SerializableFunction extends Function, Serializable {
+}
diff --git a/src/main/java/net/openhft/chronicle/hash/function/package-info.java b/src/main/java/net/openhft/chronicle/hash/function/package-info.java
index 8607e8247..c3b318ff6 100644
--- a/src/main/java/net/openhft/chronicle/hash/function/package-info.java
+++ b/src/main/java/net/openhft/chronicle/hash/function/package-info.java
@@ -15,7 +15,6 @@
*/
/**
- * Copies of some interfaces from Java 8 used by Chronicle Map project
- * for Java 7+.
+ * Serializable versions of functions for convenience of anonymous implementation
*/
package net.openhft.chronicle.hash.function;
\ No newline at end of file
diff --git a/src/main/java/net/openhft/chronicle/map/BigSegmentHeader.java b/src/main/java/net/openhft/chronicle/hash/impl/BigSegmentHeader.java
similarity index 96%
rename from src/main/java/net/openhft/chronicle/map/BigSegmentHeader.java
rename to src/main/java/net/openhft/chronicle/hash/impl/BigSegmentHeader.java
index 48c4a5f8a..e02488138 100644
--- a/src/main/java/net/openhft/chronicle/map/BigSegmentHeader.java
+++ b/src/main/java/net/openhft/chronicle/hash/impl/BigSegmentHeader.java
@@ -14,7 +14,7 @@
* limitations under the License.
*/
-package net.openhft.chronicle.map;
+package net.openhft.chronicle.hash.impl;
import net.openhft.chronicle.hash.locks.IllegalInterProcessLockStateException;
@@ -25,8 +25,8 @@ import static java.nio.ByteOrder.nativeOrder;
import static net.openhft.lang.io.AbstractBytes.UNSIGNED_INT_MASK;
import static net.openhft.lang.io.NativeBytes.UNSAFE;
-enum BigSegmentHeader implements SegmentHeader {
- INSTANCE;
+public final class BigSegmentHeader implements SegmentHeader {
+ public static final BigSegmentHeader INSTANCE = new BigSegmentHeader();
static final long LOCK_OFFSET = 0L; // 64-bit
static final long COUNT_WORD_OFFSET = LOCK_OFFSET;
@@ -51,8 +51,13 @@ enum BigSegmentHeader implements SegmentHeader {
static final long SIZE_OFFSET = LOCK_OFFSET + 8L; // 32-bit
static final long NEXT_POS_TO_SEARCH_FROM_OFFSET = SIZE_OFFSET + 4L;
+
static final long EXCLUSIVE_LOCK_HOLDER_THREAD_ID_OFFSET = NEXT_POS_TO_SEARCH_FROM_OFFSET + 4L;
+ static final long DELETED_OFFSET = EXCLUSIVE_LOCK_HOLDER_THREAD_ID_OFFSET + 8L;
+
+ private BigSegmentHeader() {}
+
@Override
public long size(long address) {
return UNSAFE.getInt(address + SIZE_OFFSET) & UNSIGNED_INT_MASK;
@@ -63,6 +68,16 @@ enum BigSegmentHeader implements SegmentHeader {
UNSAFE.putInt(address + SIZE_OFFSET, (int) size);
}
+ @Override
+ public long deleted(long address) {
+ return UNSAFE.getInt(address + DELETED_OFFSET) & UNSIGNED_INT_MASK;
+ }
+
+ @Override
+ public void deleted(long address, long deleted) {
+ UNSAFE.putInt(address + DELETED_OFFSET, (int) deleted);
+ }
+
@Override
public long nextPosToSearchFrom(long address) {
return UNSAFE.getInt(address + NEXT_POS_TO_SEARCH_FROM_OFFSET) & UNSIGNED_INT_MASK;
diff --git a/src/main/java/net/openhft/chronicle/hash/impl/ChronicleHashBuilderImpl.java b/src/main/java/net/openhft/chronicle/hash/impl/ChronicleHashBuilderImpl.java
new file mode 100644
index 000000000..78e721cfd
--- /dev/null
+++ b/src/main/java/net/openhft/chronicle/hash/impl/ChronicleHashBuilderImpl.java
@@ -0,0 +1,43 @@
+/*
+ * 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.hash.impl;
+
+
+import net.openhft.chronicle.hash.ChronicleHash;
+import net.openhft.chronicle.hash.ChronicleHashBuilder;
+import net.openhft.chronicle.hash.serialization.internal.SerializationBuilder;
+
+//TODO remove this temporary interface
+public interface ChronicleHashBuilderImpl,
+ B extends ChronicleHashBuilder> extends ChronicleHashBuilder {
+
+ SerializationBuilder keyBuilder();
+
+ int actualSegments(boolean replicated);
+
+ long entriesPerSegment(boolean replicated);
+
+ long chunkSize(boolean replicated);
+
+ int segmentHeaderSize(boolean replicated);
+
+ long actualChunksPerSegment(boolean replicated);
+
+ int maxChunksPerEntry();
+
+ int segmentEntrySpaceInnerOffset(boolean replicated);
+}
diff --git a/src/main/java/net/openhft/chronicle/hash/impl/ContextFactory.java b/src/main/java/net/openhft/chronicle/hash/impl/ContextFactory.java
new file mode 100644
index 000000000..67827a764
--- /dev/null
+++ b/src/main/java/net/openhft/chronicle/hash/impl/ContextFactory.java
@@ -0,0 +1,26 @@
+/*
+ * 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.hash.impl;
+
+public interface ContextFactory {
+
+ T createContext(HashContext root, int indexInContextCache);
+
+ T createRootContext();
+
+ Class contextClass();
+}
diff --git a/src/main/java/net/openhft/chronicle/hash/impl/HashContext.java b/src/main/java/net/openhft/chronicle/hash/impl/HashContext.java
new file mode 100644
index 000000000..d55dc9769
--- /dev/null
+++ b/src/main/java/net/openhft/chronicle/hash/impl/HashContext.java
@@ -0,0 +1,1503 @@
+/*
+ * 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.hash.impl;
+
+import net.openhft.chronicle.hash.ChronicleHash;
+import net.openhft.chronicle.hash.KeyContext;
+import net.openhft.chronicle.hash.hashing.LongHashFunction;
+import net.openhft.chronicle.hash.impl.hashlookup.HashLookup;
+import net.openhft.chronicle.hash.locks.IllegalInterProcessLockStateException;
+import net.openhft.chronicle.hash.locks.InterProcessLock;
+import net.openhft.chronicle.hash.serialization.BytesReader;
+import net.openhft.chronicle.hash.serialization.internal.BytesBytesInterop;
+import net.openhft.chronicle.hash.serialization.internal.DelegatingMetaBytesInterop;
+import net.openhft.chronicle.hash.serialization.internal.MetaBytesInterop;
+import net.openhft.lang.collection.DirectBitSet;
+import net.openhft.lang.collection.SingleThreadedDirectBitSet;
+import net.openhft.lang.io.ByteBufferBytes;
+import net.openhft.lang.io.Bytes;
+import net.openhft.lang.io.MultiStoreBytes;
+import net.openhft.lang.threadlocal.ThreadLocalCopies;
+import org.jetbrains.annotations.NotNull;
+
+import java.lang.ref.Reference;
+import java.lang.ref.ReferenceQueue;
+import java.lang.ref.WeakReference;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.ConcurrentModificationException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Condition;
+
+import static net.openhft.chronicle.hash.impl.HashContext.SearchState.ABSENT;
+import static net.openhft.chronicle.hash.impl.HashContext.SearchState.DELETED;
+import static net.openhft.chronicle.hash.impl.HashContext.SearchState.PRESENT;
+
+public abstract class HashContext>
+ implements KeyContext {
+
+
+ /////////////////////////////////////////////////
+ // Inner state & lifecycle
+ static final ThreadLocal threadLocalContextCache =
+ new ThreadLocal<>();
+
+ public static T get(ContextFactory contextFactory) {
+ HashContext cache = threadLocalContextCache.get();
+ if (cache != null)
+ return (T) cache.getContext(contextFactory);
+ T rootContext = contextFactory.createRootContext();
+ threadLocalContextCache.set(rootContext);
+ return rootContext;
+ }
+
+ final ArrayList contexts;
+ final ReferenceQueue hashAndContextLocalsQueue;
+ final Thread owner;
+ public final ThreadLocalCopies copies;
+ final HashContext contextCache;
+ final int indexInContextCache;
+ boolean used;
+
+ public HashContext() {
+ contexts = new ArrayList<>(1);
+ contextCache = this;
+ indexInContextCache = 0;
+ contexts.add(this);
+ hashAndContextLocalsQueue = new ReferenceQueue<>();
+ owner = Thread.currentThread();
+ copies = new ThreadLocalCopies();
+ used = true;
+ }
+
+ public HashContext(HashContext contextCache, int indexInContextCache) {
+ contexts = null;
+ hashAndContextLocalsQueue = null;
+ owner = Thread.currentThread();
+ copies = new ThreadLocalCopies();
+ this.contextCache = contextCache;
+ this.indexInContextCache = indexInContextCache;
+ }
+
+ T getContext(ContextFactory contextFactory) {
+ for (HashContext context : contexts) {
+ if (context.getClass() == contextFactory.contextClass() &&
+ !context.used) {
+ context.used = true;
+ return (T) context;
+ }
+ }
+ int maxNestedContexts = 1 << 16;
+ if (contexts.size() > maxNestedContexts) {
+ throw new IllegalStateException("More than " + maxNestedContexts +
+ " nested ChronicleHash contexts are not supported. Very probable that " +
+ "you simply forgot to close context somewhere (recommended to use " +
+ "try-with-resources statement). " +
+ "Otherwise this is a bug, please report with this " +
+ "stack trace on https://github.com/OpenHFT/Chronicle-Map/issues");
+ }
+ T context = contextFactory.createContext(this, contexts.size());
+ context.used = true;
+ contexts.add(context);
+ return context;
+ }
+
+ public boolean used() {
+ return used;
+ }
+
+ public void checkOnEachPublicOperation() {
+ if (owner != Thread.currentThread()) {
+ throw new ConcurrentModificationException(
+ "Context shouldn't be accessed from multiple threads");
+ }
+ if (!used)
+ throw new IllegalStateException("Context shouldn't be accessed after close()");
+ }
+
+ @Override
+ public void close() {
+ checkOnEachPublicOperation();
+ doClose();
+ }
+
+ public void doClose() {
+ closeHash();
+ used = false;
+ totalCheckClosed();
+ }
+
+ public void totalCheckClosed() {
+ assert !hashInit() : "map not closed";
+ assert !hashAndContextLocalsInit() : "map&context locals not closed";
+ assert !keyModelInit() : "key model not closed";
+ assert !keyReaderInit() : "key reader not closed";
+ assert !keyInit() : "key not closed";
+ assert !keyHashInit() : "key hash not closed";
+ assert !segmentIndexInit() : "segment index not closed";
+ assert !segmentHeaderInit() : "segment header not closed";
+ assert !locksInit() : "locks not closed";
+ assert !segmentInit() : "segment not closed";
+ assert !hashLookupInit() : "hash lookup not closed";
+ assert !keySearchInit() : "key search not closed";
+ assert !entrySizeInChunksInit() : "entry size in chunks not closed";
+ }
+
+
+ /////////////////////////////////////////////////
+ // Hash
+ public VanillaChronicleHash h;
+
+ public boolean hashInit() {
+ return h != null;
+ }
+
+ public void initHash(VanillaChronicleHash hash) {
+ initHashDependencies();
+ initHash0(hash);
+ }
+
+ public void initHashDependencies() {
+ // no dependencies
+ }
+
+ void initHash0(VanillaChronicleHash hash) {
+ h = hash;
+ }
+
+ public void checkHashInit() {
+ if (!hashInit())
+ throw new IllegalStateException("Hash should be init");
+ }
+
+ public void closeHash() {
+ if (!hashInit())
+ return;
+ closeHashDependants();
+ closeHash0();
+ }
+
+ public void closeHashDependants() {
+ // TODO update dependants
+ closeHashAndContextLocals();
+ closeSegmentHeader();
+ closeKeyModel();
+ closeKey();
+ closeKeyReader();
+ closeSegmentIndex();
+ closeLocks();
+ }
+
+ void closeHash0() {
+ h = null;
+ }
+
+
+ /////////////////////////////////////////////////
+ // Hash and context locals
+ final ArrayList> hashAndContextLocalsCache =
+ new ArrayList<>(1);
+ HashAndContextLocalsReference hashAndContextLocalsReference;
+
+ public static class HashAndContextLocals {
+ K reusableKey;
+ public HashAndContextLocals(ChronicleHash hash) {
+ // TODO why not using hash.newKeyInstance()
+ if (hash.keyClass() == CharSequence.class)
+ reusableKey = (K) new StringBuilder();
+ }
+ }
+
+ static class HashAndContextLocalsReference
+ extends WeakReference> {
+ HashAndContextLocals hashAndContextLocals;
+ public HashAndContextLocalsReference(
+ ChronicleHash hash, ReferenceQueue hashAndContextLocalsQueue,
+ HashAndContextLocals hashAndContextLocals) {
+ super(hash, hashAndContextLocalsQueue);
+ this.hashAndContextLocals = hashAndContextLocals;
+ }
+ }
+
+ public boolean hashAndContextLocalsInit() {
+ return hashAndContextLocalsReference != null;
+ }
+
+ public void initHashAndContextLocals() {
+ if (hashAndContextLocalsInit())
+ return;
+ initHashAndContextLocalsDependencies();
+ initHashAndContextLocals0();
+ }
+
+ public void initHashAndContextLocalsDependencies() {
+ checkHashInit();
+ }
+
+ void initHashAndContextLocals0() {
+ pollLocalsQueue();
+ int firstEmptyIndex = -1;
+ int cacheSize = hashAndContextLocalsCache.size();
+ for (int i = 0; i < cacheSize; i++) {
+ HashAndContextLocalsReference ref = hashAndContextLocalsCache.get(i);
+ ChronicleHash referencedHash;
+ if (ref == null || (referencedHash = ref.get()) == null) {
+ if (firstEmptyIndex < 0)
+ firstEmptyIndex = i;
+ if (ref != null)
+ hashAndContextLocalsCache.set(i, null);
+ continue;
+ }
+ if (referencedHash == h) {
+ hashAndContextLocalsReference = ref;
+ return;
+ }
+ }
+ HashAndContextLocals hashAndContextLocals = newHashAndContextLocals();
+ int indexToInsert = firstEmptyIndex < 0 ? cacheSize : firstEmptyIndex;
+ HashAndContextLocalsReference ref = new HashAndContextLocalsReference<>(h,
+ contextCache.hashAndContextLocalsQueue, hashAndContextLocals);
+ hashAndContextLocalsReference = ref;
+ hashAndContextLocalsCache.add(indexToInsert, ref);
+ }
+
+ public HashAndContextLocals newHashAndContextLocals() {
+ return new HashAndContextLocals<>(h);
+ }
+
+ public HashAndContextLocals hashAndContextLocals() {
+ return hashAndContextLocalsReference.hashAndContextLocals;
+ }
+
+ void pollLocalsQueue() {
+ Reference extends ChronicleHash> ref;
+ while ((ref = contextCache.hashAndContextLocalsQueue.poll()) != null) {
+ // null object pointing to potentially large cached instances, for GC
+ ((HashAndContextLocalsReference) ref).hashAndContextLocals = null;
+ }
+ }
+
+ public void closeHashAndContextLocals() {
+ if (!hashAndContextLocalsInit())
+ return;
+ closeHashContextAndLocalsDependants();
+ closeHashAndContextLocals0();
+ }
+
+ public void closeHashContextAndLocalsDependants() {
+ // TODO determine dependencies
+ }
+
+ void closeHashAndContextLocals0() {
+ hashAndContextLocalsReference = null;
+ }
+
+
+ /////////////////////////////////////////////////
+ // Key model
+ KI keyInterop;
+
+ public void initKeyModel() {
+ if (keyModelInit())
+ return;
+ initKeyModelDependencies();
+ initKeyModel0();
+ }
+
+ public boolean keyModelInit() {
+ return keyInterop != null;
+ }
+
+ public void initKeyModelDependencies() {
+ checkHashInit();
+ }
+
+ public void initKeyModel0() {
+ keyInterop = h.keyInteropProvider.get(copies, h.originalKeyInterop);
+ }
+
+ public void closeKeyModel() {
+ if (!keyModelInit())
+ return;
+ closeKeyModelDependants();
+ closeKeyModel0();
+ }
+
+ public void closeKeyModelDependants() {
+ closeKey();
+ }
+
+ void closeKeyModel0() {
+ keyInterop = null;
+ }
+
+
+ /////////////////////////////////////////////////
+ // Key reader
+ BytesReader keyReader;
+
+ public void initKeyReader() {
+ if (keyReaderInit())
+ return;
+ initKeyReaderDependencies();
+ initKeyReader0();
+ }
+
+ public boolean keyReaderInit() {
+ return keyReader != null;
+ }
+
+ public void initKeyReaderDependencies() {
+ checkHashInit();
+ }
+
+ void initKeyReader0() {
+ keyReader = h.keyReaderProvider.get(copies, h.originalKeyReader);
+ }
+
+ public void closeKeyReader() {
+ if (!keyReaderInit())
+ return;
+ closeKeyReaderDependants();
+ closeKeyReader0();
+ }
+
+ public void closeKeyReaderDependants() {
+ closeKey();
+ }
+
+ void closeKeyReader0() {
+ keyReader = null;
+ }
+
+
+ /////////////////////////////////////////////////
+ // Key
+ K key;
+ public MKI metaKeyInterop;
+ public long keySize;
+
+ public void initKey(K key) {
+ initKeyDependencies();
+ initKey0(key);
+ }
+
+ public boolean keyInit() {
+ return key != null;
+ }
+
+ public void initKeyDependencies() {
+ initKeyModel();
+ }
+
+ public void initKey0(K key) {
+ h.checkKey(key);
+ this.key = key;
+ metaKeyInterop = h.metaKeyInteropProvider.get(
+ copies, h.originalMetaKeyInterop, keyInterop, key);
+ keySize = metaKeyInterop.size(keyInterop, key);
+ }
+
+ public void checkKeyInit() {
+ if (!keyInit())
+ throw new IllegalStateException("Key should be init");
+ }
+
+ public void closeKey() {
+ if (!keyInit())
+ return;
+ closeKeyDependants();
+ closeKey0();
+ }
+
+ public void closeKeyDependants() {
+ closeKeyHash();
+ }
+
+ public void closeKey0() {
+ key = null;
+ metaKeyInterop = null;
+ }
+
+ @Override
+ public long keySize() {
+ checkOnEachPublicOperation();
+ checkKeyInit();
+ return keySize0();
+ }
+
+ public long keySize0() {
+ return keySize;
+ }
+
+ @NotNull
+ @Override
+ public K key() {
+ if (keyInit())
+ return key;
+ initKeySearch();
+ initHashAndContextLocals();
+ hashAndContextLocals().reusableKey = key0(hashAndContextLocals().reusableKey);
+ assert key != null;
+ return key;
+ }
+
+
+ /////////////////////////////////////////////////
+ // Key hash
+ long hash;
+
+ public void initKeyHash() {
+ if (keyHashInit())
+ return;
+ initKeyHashDependencies();
+ initKeyHash0();
+ }
+
+ public boolean keyHashInit() {
+ return hash != 0;
+ }
+
+ public void initKeyHashDependencies() {
+ checkKeyInit();
+ initKeyModel();
+ }
+
+ void initKeyHash0() {
+ hash = metaKeyInterop.hash(keyInterop, LongHashFunction.city_1_1(), key);
+ }
+
+ public void closeKeyHash() {
+ // don't skip closing if hash = 0 because this is a valid hash also
+ closeKeyHashDependants();
+ closeKeyHash0();
+ }
+
+ public void closeKeyHashDependants() {
+ closeSegmentIndex();
+ }
+
+ void closeKeyHash0() {
+ hash = 0L;
+ }
+
+
+ /////////////////////////////////////////////////
+ // Segment index
+ public int segmentIndex = -1;
+
+ public void initSegmentIndex() {
+ if (segmentIndexInit())
+ return;
+ initSegmentIndexDependencies();
+ initSegmentIndex0();
+ }
+
+ public boolean segmentIndexInit() {
+ return segmentIndex >= 0;
+ }
+
+ public void initSegmentIndexDependencies() {
+ initKeyHash();
+ }
+
+ void initSegmentIndex0() {
+ segmentIndex = h.hashSplitting.segmentIndex(hash);
+ }
+
+ public void closeSegmentIndex() {
+ if (!segmentIndexInit())
+ return;
+ closeSegmentIndexDependants();
+ closeSegmentIndex0();
+ }
+
+ public void closeSegmentIndexDependants() {
+ closeSegmentHeader();
+ }
+
+ void closeSegmentIndex0() {
+ segmentIndex = -1;
+ }
+
+
+ /////////////////////////////////////////////////
+ // Segment header
+ long segmentHeaderAddress;
+ public SegmentHeader segmentHeader;
+
+ public void initSegmentHeader() {
+ if (segmentHeaderInit())
+ return;
+ initSegmentHeaderDependencies();
+ initSegmentHeader0();
+ }
+
+ public boolean segmentHeaderInit() {
+ return segmentHeader != null;
+ }
+
+ public void initSegmentHeaderDependencies() {
+ initSegmentIndex();
+ }
+
+ void initSegmentHeader0() {
+ segmentHeaderAddress = h.ms.address() + h.segmentHeaderOffset(segmentIndex);
+ segmentHeader = BigSegmentHeader.INSTANCE;
+ }
+
+ public void closeSegmentHeader() {
+ if (!segmentHeaderInit())
+ return;
+ closeSegmentHeaderDependants();
+ closeSegmentHeader0();
+ }
+
+ public void closeSegmentHeaderDependants() {
+ closeLocks();
+ closeSegment();
+ }
+
+ void closeSegmentHeader0() {
+ segmentHeader = null;
+ }
+
+ public long entries() {
+ initSegmentHeader();
+ return segmentHeader.size(segmentHeaderAddress);
+ }
+
+ public void entries(long size) {
+ segmentHeader.size(segmentHeaderAddress, size);
+ }
+
+ public long nextPosToSearchFrom() {
+ return segmentHeader.nextPosToSearchFrom(segmentHeaderAddress);
+ }
+
+ public void nextPosToSearchFrom(long nextPosToSearchFrom) {
+ segmentHeader.nextPosToSearchFrom(segmentHeaderAddress, nextPosToSearchFrom);
+ }
+
+ public long deleted() {
+ initSegmentHeader();
+ return segmentHeader.deleted(segmentHeaderAddress);
+ }
+
+ public void deleted(long deleted) {
+ segmentHeader.deleted(segmentHeaderAddress, deleted);
+ }
+
+ public long size() {
+ return entries() - deleted();
+ }
+
+
+ /////////////////////////////////////////////////
+ // Locks
+ HashContext rootContextOnThisSegment;
+ int readLockCount;
+ int updateLockCount;
+ public int writeLockCount;
+ int totalReadLockCount;
+ int totalUpdateLockCount;
+ int totalWriteLockCount;
+
+ final InterProcessLock readLock = new ReadLock();
+ final InterProcessLock updateLock = new UpdateLock();
+ final InterProcessLock writeLock = new WriteLock();
+
+ public void initLocks() {
+ if (locksInit())
+ return;
+ initLocksDependencies();
+ initLocks0();
+ }
+
+ public boolean locksInit() {
+ return rootContextOnThisSegment != null;
+ }
+
+ public void initLocksDependencies() {
+ initSegmentHeader();
+ }
+
+ void initLocks0() {
+ readLockCount = 0;
+ updateLockCount = 0;
+ writeLockCount = 0;
+ for (int i = 0; i < indexInContextCache; i++) {
+ HashContext parentContext = contextCache.contexts.get(i);
+ if (parentContext.segmentHeader != null &&
+ parentContext.segmentHeaderAddress == segmentHeaderAddress) {
+ rootContextOnThisSegment = parentContext;
+ return;
+ }
+ }
+ rootContextOnThisSegment = this;
+ totalReadLockCount = 0;
+ totalUpdateLockCount = 0;
+ totalWriteLockCount = 0;
+ }
+
+ public void closeLocks() {
+ if (!locksInit())
+ return;
+ closeLocksDependants();
+ closeLocks0();
+ }
+
+ public void closeLocksDependants() {
+ // TODO determine
+ closeKeySearch();
+ }
+
+ void closeLocks0() {
+ // TODO should throw ISE on attempt to close root context when children not closed
+ if (rootContextOnThisSegment == this) {
+ if (totalWriteLockCount > 0) {
+ segmentHeader.writeUnlock(segmentHeaderAddress);
+ closeKeySearch();
+ } else if (totalUpdateLockCount > 0) {
+ segmentHeader.updateUnlock(segmentHeaderAddress);
+ closeKeySearch();
+ } else if (totalReadLockCount > 0) {
+ segmentHeader.readUnlock(segmentHeaderAddress);
+ closeKeySearch();
+ }
+ } else {
+ if (writeLockCount > 0 &&
+ rootContextOnThisSegment.totalReadLockCount == writeLockCount) {
+ if (shouldUpdateUnlock()) {
+ if (shouldReadUnlock()) {
+ segmentHeader.writeUnlock(segmentHeaderAddress);
+ closeKeySearch();
+ } else {
+ segmentHeader.downgradeWriteToReadLock(segmentHeaderAddress);
+ }
+ } else {
+ segmentHeader.downgradeWriteToUpdateLock(segmentHeaderAddress);
+ }
+ } else if (shouldUpdateUnlock()) {
+ if (shouldReadUnlock()) {
+ segmentHeader.updateUnlock(segmentHeaderAddress);
+ closeKeySearch();
+ } else {
+ segmentHeader.downgradeUpdateToReadLock(segmentHeaderAddress);
+ }
+ } else if (shouldReadUnlock()) {
+ segmentHeader.readUnlock(segmentHeaderAddress);
+ closeKeySearch();
+ }
+ }
+ rootContextOnThisSegment.totalReadLockCount -= readLockCount;
+ rootContextOnThisSegment.totalUpdateLockCount -= updateLockCount;
+ rootContextOnThisSegment.totalWriteLockCount -= writeLockCount;
+ readLockCount = updateLockCount = writeLockCount = 0;
+ rootContextOnThisSegment = null;
+ }
+
+ public boolean isReadLocked() {
+ return rootContextOnThisSegment.totalReadLockCount > 0;
+ }
+
+ public boolean isUpdateLocked() {
+ return rootContextOnThisSegment.totalUpdateLockCount > 0;
+ }
+
+ public boolean isWriteLocked() {
+ return rootContextOnThisSegment.totalWriteLockCount > 0;
+ }
+
+ public void upgradeToWriteLock() {
+ if (!isWriteLocked())
+ writeLock();
+ }
+
+ boolean shouldReadUnlock() {
+ return readLockCount > 0 &&
+ rootContextOnThisSegment.totalReadLockCount == readLockCount;
+ }
+
+ boolean shouldUpdateUnlock() {
+ return updateLockCount > 0 &&
+ rootContextOnThisSegment.totalUpdateLockCount == updateLockCount;
+ }
+
+ void incrementReadCounts() {
+ rootContextOnThisSegment.totalReadLockCount++;
+ readLockCount++;
+ }
+
+ void decrementReadCounts() {
+ rootContextOnThisSegment.totalReadLockCount--;
+ readLockCount--;
+ }
+
+ void incrementUpdateCounts() {
+ incrementReadCounts();
+ rootContextOnThisSegment.totalUpdateLockCount++;
+ updateLockCount++;
+ }
+
+ void decrementUpdateCounts() {
+ decrementReadCounts();
+ rootContextOnThisSegment.totalUpdateLockCount--;
+ updateLockCount--;
+ }
+
+ void incrementWriteCounts() {
+ incrementUpdateCounts();
+ rootContextOnThisSegment.totalWriteLockCount++;
+ writeLockCount++;
+ }
+
+ void decrementWriteCounts() {
+ decrementUpdateCounts();
+ rootContextOnThisSegment.totalWriteLockCount--;
+ writeLockCount--;
+ }
+
+ abstract class AbstractLock implements InterProcessLock {
+ abstract boolean fastLock();
+ abstract void doLock();
+ abstract void doLockInterruptibly();
+ abstract boolean doTryLock(long time, TimeUnit unit);
+ abstract void incrementCounts();
+
+ @Override
+ public void lock() {
+ checkOnEachPublicOperation();
+ if (!fastLock())
+ doLock();
+ incrementCounts();
+ }
+
+ @Override
+ public void lockInterruptibly() throws InterruptedException {
+ checkOnEachPublicOperation();
+ if (!fastLock())
+ doLockInterruptibly();
+ incrementCounts();
+ }
+
+ @Override
+ public boolean tryLock(long time, @NotNull TimeUnit unit) throws InterruptedException {
+ checkOnEachPublicOperation();
+ if (fastLock() || doTryLock(time, unit)) {
+ incrementCounts();
+ return true;
+ } else {
+ return false;
+ }
+ }
+
+ @NotNull
+ @Override
+ public Condition newCondition() {
+ throw new UnsupportedOperationException();
+ }
+ }
+
+ class ReadLock extends AbstractLock {
+ @Override
+ boolean fastLock() {
+ return isReadLocked();
+ }
+
+ @Override
+ public boolean isHeldByCurrentThread() {
+ return fastLock();
+ }
+
+ @Override
+ void doLock() {
+ segmentHeader.readLock(segmentHeaderAddress);
+ }
+
+ @Override
+ void doLockInterruptibly() {
+ segmentHeader.readLockInterruptibly(segmentHeaderAddress);
+ }
+
+ @Override
+ public boolean tryLock() {
+ checkOnEachPublicOperation();
+ if (fastLock() || segmentHeader.tryReadLock(segmentHeaderAddress)) {
+ incrementCounts();
+ return true;
+ } else {
+ return false;
+ }
+ }
+
+ @Override
+ boolean doTryLock(long time, TimeUnit unit) {
+ return segmentHeader.tryReadLock(segmentHeaderAddress, time, unit);
+ }
+
+ @Override
+ void incrementCounts() {
+ incrementReadCounts();
+ }
+
+ @Override
+ public void unlock() {
+ if (readLockCount == 0)
+ throw new IllegalInterProcessLockStateException("Read lock is not held");
+ if (rootContextOnThisSegment.totalReadLockCount == 1) {
+ segmentHeader.readUnlock(segmentHeaderAddress);
+ closeKeySearch();
+ }
+ decrementReadCounts();
+ }
+ }
+
+ class UpdateLock extends AbstractLock {
+ @Override
+ boolean fastLock() {
+ if (rootContextOnThisSegment.totalUpdateLockCount == 0) {
+ if (rootContextOnThisSegment.totalReadLockCount != 0) {
+ throw new IllegalInterProcessLockStateException("Must not acquire update " +
+ "lock, while read lock is already held by this thread");
+ }
+ return false;
+ } else {
+ return true;
+ }
+ }
+
+ @Override
+ public boolean isHeldByCurrentThread() {
+ return isUpdateLocked();
+ }
+
+ @Override
+ void doLock() {
+ segmentHeader.updateLock(segmentHeaderAddress);
+ }
+
+ @Override
+ void doLockInterruptibly() {
+ segmentHeader.updateLockInterruptibly(segmentHeaderAddress);
+ }
+
+ @Override
+ public boolean tryLock() {
+ checkOnEachPublicOperation();
+ if (doTryLock()) {
+ incrementCounts();
+ return true;
+ } else {
+ return false;
+ }
+ }
+
+ boolean doTryLock() {
+ if (rootContextOnThisSegment.totalUpdateLockCount == 0) {
+ if (rootContextOnThisSegment.totalReadLockCount > 0) {
+ return segmentHeader.tryUpgradeReadToUpdateLock(segmentHeaderAddress);
+ } else {
+ return segmentHeader.tryUpdateLock(segmentHeaderAddress);
+ }
+ } else {
+ return true;
+ }
+ }
+
+ @Override
+ boolean doTryLock(long time, TimeUnit unit) {
+ return segmentHeader.tryUpdateLock(segmentHeaderAddress, time, unit);
+ }
+
+ @Override
+ void incrementCounts() {
+ incrementUpdateCounts();
+ }
+
+ @Override
+ public void unlock() {
+ if (updateLockCount == 0)
+ throw new IllegalInterProcessLockStateException("Update lock is not held");
+ if (rootContextOnThisSegment.totalUpdateLockCount == 1) {
+ if (rootContextOnThisSegment.totalReadLockCount == 1) {
+ segmentHeader.updateUnlock(segmentHeaderAddress);
+ closeKeySearch();
+ } else {
+ segmentHeader.downgradeUpdateToReadLock(segmentHeaderAddress);
+ }
+ }
+ decrementUpdateCounts();
+ }
+ }
+
+ class WriteLock extends AbstractLock {
+ @Override
+ boolean fastLock() {
+ if (rootContextOnThisSegment.totalWriteLockCount == 0) {
+ if (rootContextOnThisSegment.totalUpdateLockCount == 0 &&
+ rootContextOnThisSegment.totalReadLockCount != 0) {
+ throw new IllegalInterProcessLockStateException("Must not acquire write " +
+ "lock, while read lock is already held by this thread");
+ }
+ return false;
+ } else {
+ return true;
+ }
+ }
+
+ @Override
+ public boolean isHeldByCurrentThread() {
+ return isWriteLocked();
+ }
+
+ @Override
+ void doLock() {
+ if (rootContextOnThisSegment.totalUpdateLockCount > 0) {
+ segmentHeader.upgradeUpdateToWriteLock(segmentHeaderAddress);
+ } else {
+ segmentHeader.writeLock(segmentHeaderAddress);
+ }
+ }
+
+ @Override
+ void doLockInterruptibly() {
+ if (rootContextOnThisSegment.totalUpdateLockCount > 0) {
+ segmentHeader.upgradeUpdateToWriteLockInterruptibly(segmentHeaderAddress);
+ } else {
+ segmentHeader.writeLockInterruptibly(segmentHeaderAddress);
+ }
+ }
+
+ @Override
+ public boolean tryLock() {
+ checkOnEachPublicOperation();
+ if (doTryLock()) {
+ incrementCounts();
+ return true;
+ } else {
+ return false;
+ }
+ }
+
+ boolean doTryLock() {
+ if (rootContextOnThisSegment.totalWriteLockCount == 0) {
+ if (rootContextOnThisSegment.totalUpdateLockCount > 0) {
+ return segmentHeader.tryUpgradeUpdateToWriteLock(segmentHeaderAddress);
+ } else if (rootContextOnThisSegment.totalReadLockCount > 0) {
+ return segmentHeader.tryUpgradeReadToWriteLock(segmentHeaderAddress);
+ } else {
+ return segmentHeader.tryWriteLock(segmentHeaderAddress);
+ }
+ } else {
+ return true;
+ }
+ }
+
+ @Override
+ boolean doTryLock(long time, TimeUnit unit) {
+ if (rootContextOnThisSegment.totalUpdateLockCount > 0) {
+ return segmentHeader.tryUpgradeUpdateToWriteLock(
+ segmentHeaderAddress, time, unit);
+ } else {
+ return segmentHeader.tryWriteLock(segmentHeaderAddress, time, unit);
+ }
+ }
+
+ @Override
+ void incrementCounts() {
+ incrementWriteCounts();
+ }
+
+ @Override
+ public void unlock() {
+ if (writeLockCount == 0)
+ throw new IllegalInterProcessLockStateException("Write lock is not held");
+ if (rootContextOnThisSegment.totalWriteLockCount == 1) {
+ if (rootContextOnThisSegment.totalUpdateLockCount == 1) {
+ if (rootContextOnThisSegment.totalReadLockCount == 1) {
+ segmentHeader.writeUnlock(segmentHeaderAddress);
+ closeKeySearch();
+ } else {
+ segmentHeader.downgradeWriteToReadLock(segmentHeaderAddress);
+ }
+ } else {
+ segmentHeader.downgradeWriteToUpdateLock(segmentHeaderAddress);
+ }
+ }
+ decrementWriteCounts();
+ }
+ }
+
+ @NotNull
+ @Override
+ public InterProcessLock readLock() {
+ checkOnEachPublicOperation();
+ initLocks();
+ return readLock;
+ }
+
+ @NotNull
+ @Override
+ public InterProcessLock updateLock() {
+ checkOnEachPublicOperation();
+ initLocks();
+ return updateLock;
+ }
+
+ @NotNull
+ @Override
+ public InterProcessLock writeLock() {
+ checkOnEachPublicOperation();
+ initLocks();
+ return writeLock;
+ }
+
+
+ /////////////////////////////////////////////////
+ // Segment
+ public final HashLookup hashLookup = new HashLookup();
+ final MultiStoreBytes freeListBytes = new MultiStoreBytes();
+ public final SingleThreadedDirectBitSet freeList = new SingleThreadedDirectBitSet();
+ long entrySpaceOffset;
+
+ public void initSegment() {
+ if (segmentInit())
+ return;
+ initSegmentDependencies();
+ initSegment0();
+ }
+
+ public boolean segmentInit() {
+ return entrySpaceOffset != 0;
+ }
+
+ public void initSegmentDependencies() {
+ initSegmentHeader();
+ }
+
+ void initSegment0() {
+ long hashLookupOffset = h.segmentOffset(segmentIndex);
+ hashLookup.reuse(h.ms.address() + hashLookupOffset,
+ h.segmentHashLookupCapacity, h.segmentHashLookupEntrySize,
+ h.segmentHashLookupKeyBits, h.segmentHashLookupValueBits);
+ long freeListOffset = hashLookupOffset + h.segmentHashLookupOuterSize;
+ freeListBytes.storePositionAndSize(h.ms, freeListOffset, h.segmentFreeListInnerSize);
+ freeList.reuse(freeListBytes);
+ entrySpaceOffset = freeListOffset + h.segmentFreeListOuterSize +
+ h.segmentEntrySpaceInnerOffset;
+ }
+
+ public void closeSegment() {
+ if (!segmentInit())
+ return;
+ closeSegmentDependants();
+ closeSegment0();
+ }
+
+ public void closeSegmentDependants() {
+ closeHashLookup();
+ }
+
+ void closeSegment0() {
+ entrySpaceOffset = 0L;
+ }
+
+
+ /////////////////////////////////////////////////
+ // Hash lookup
+ public void initHashLookup() {
+ if (hashLookupInit())
+ return;
+ initHashLookupDependencies();
+ hashLookup.init0(h.hashSplitting.segmentHash(hash));
+ }
+
+ public boolean hashLookupInit() {
+ return hashLookup.isInit();
+ }
+
+ public void initHashLookupDependencies() {
+ initSegment();
+ }
+
+ public void closeHashLookup() {
+ if (!hashLookupInit())
+ return;
+ closeHashLookupDependants();
+ hashLookup.close0();
+ }
+
+ public void closeHashLookupDependants() {
+ closeKeySearch();
+ }
+
+
+ /////////////////////////////////////////////////
+ // Entry
+ final MultiStoreBytes entryCache = new MultiStoreBytes();
+ public MultiStoreBytes entry;
+
+ final MultiStoreBytes reuse(MultiStoreBytes entry, long pos) {
+ long offsetWithinEntrySpace = pos * h.chunkSize;
+ entry.setBytesOffset(h.bytes, entrySpaceOffset + offsetWithinEntrySpace);
+ entry.limit(h.segmentEntrySpaceInnerSize - offsetWithinEntrySpace);
+ return entry;
+ }
+
+ public final void reuse(long pos) {
+ entry = reuse(entryCache, pos);
+ }
+
+
+ /////////////////////////////////////////////////
+ // Key search, key bytes and entry
+ public static enum SearchState {
+ PRESENT,
+ DELETED,
+ ABSENT
+ }
+
+ SearchState searchState;
+ public long pos;
+ long keyOffset;
+
+ public void initKeySearch() {
+ if (keySearchInit())
+ return;
+ initKeySearchDependencies();
+ initKeySearch0();
+ }
+
+ public boolean keySearchInit() {
+ return searchState != null;
+ }
+
+ public void initKeySearchDependencies() {
+ initHashLookup();
+ initLocks();
+ if (!isReadLocked())
+ readLock().lock();
+ }
+
+ void initKeySearch0() {
+ hashLookup.initSearch0();
+ while ((pos = hashLookup.nextPos()) >= 0L) {
+ reuse(pos);
+ if (!keyEquals())
+ continue;
+ hashLookup.found();
+ initKeyOffset0();
+ keyFound();
+ return;
+ }
+ searchState = ABSENT;
+ }
+
+ boolean keyEquals() {
+ return keySize == h.keySizeMarshaller.readSize(entry) &&
+ metaKeyInterop.startsWith(keyInterop, entry, key);
+ }
+
+ public void initKeyOffset0() {
+ keyOffset = entry.position();
+ }
+
+ public void keyFound() {
+ searchState = PRESENT;
+ }
+
+ public void closeKeySearch() {
+ if (!keySearchInit())
+ return;
+ closeKeySearchDependants();
+ closeKeySearch0();
+ }
+
+ public void closeKeySearchDependants() {
+ }
+
+ void closeKeySearch0() {
+ searchState = null;
+ pos = -1;
+ entry = null;
+ hashLookup.closeSearch0();
+ }
+
+ @Override
+ public boolean containsKey() {
+ checkOnEachPublicOperation();
+ initKeySearch();
+ return containsKey0();
+ }
+
+ public boolean containsKey0() {
+ return searchStatePresent();
+ }
+
+ public SearchState searchState0() {
+ return searchState;
+ }
+
+ public boolean searchStatePresent() {
+ return searchState == PRESENT;
+ }
+
+ public void checkContainsKey() {
+ if (!containsKey())
+ throw new IllegalStateException("Key is absent");
+ }
+
+ @Override
+ public Bytes entry() {
+ checkContainsKey();
+ return entry;
+ }
+
+ @Override
+ public long keyOffset() {
+ checkContainsKey();
+ return keyOffset0();
+ }
+
+ public long keyOffset0() {
+ return keyOffset;
+ }
+
+
+ /////////////////////////////////////////////////
+ // Entry size in chunks
+ public int entrySizeInChunks;
+
+ public void initEntrySizeInChunks() {
+ if (entrySizeInChunksInit())
+ return;
+ initEntrySizeInChunksDependencies();
+ initEntrySizeInChunks0();
+ }
+
+ public boolean entrySizeInChunksInit() {
+ return entrySizeInChunks != 0;
+ }
+
+ public void initEntrySizeInChunksDependencies() {
+ }
+
+ public abstract void initEntrySizeInChunks0();
+
+ public void closeEntrySizeInChunks() {
+ if (!entrySizeInChunksInit())
+ return;
+ closeEntrySizeInChunksDependants();
+ closeEntrySizeInChunks0();
+ }
+
+ public void closeEntrySizeInChunksDependants() {
+ // TODO no dependants?
+ }
+
+ void closeEntrySizeInChunks0() {
+ entrySizeInChunks = 0;
+ }
+
+
+ /////////////////////////////////////////////////
+ // Write operations
+
+ public void updateLockIfNeeded() {
+ if (!isUpdateLocked())
+ updateLock().lock();
+ }
+
+ public int allocateEntry(long entrySize) {
+ int allocatedChunks = h.inChunks(entrySize);
+ pos = alloc(allocatedChunks);
+ reuse(pos);
+ return allocatedChunks;
+ }
+
+ public int allocateEntryAndWriteKey(long payloadSize) {
+ int allocatedChunks = allocateEntry(keySize + payloadSize);
+
+ h.keySizeMarshaller.writeSize(entry, keySize);
+ initKeyOffset0();
+ metaKeyInterop.write(keyInterop, entry, key);
+
+ if (!searchStatePresent()) {
+ // update the size before the store fence
+ entries(entries() + 1L);
+ }
+
+ return allocatedChunks;
+ }
+
+ public void commitEntryAllocation() {
+ searchState = PRESENT;
+ // put + store fence, guarantees if concurrent readers see the new entry
+ // in the hashLookup => they will also see written entry bytes
+ hashLookup.putVolatile(pos);
+ }
+
+ //TODO refactor/optimize
+ public final long alloc(int chunks) {
+ if (chunks > h.maxChunksPerEntry)
+ throw new IllegalArgumentException("Entry is too large: requires " + chunks +
+ " entry size chucks, " + h.maxChunksPerEntry + " is maximum.");
+ long ret = freeList.setNextNContinuousClearBits(nextPosToSearchFrom(), chunks);
+ if (ret == DirectBitSet.NOT_FOUND || ret + chunks > h.actualChunksPerSegment) {
+ if (ret != DirectBitSet.NOT_FOUND &&
+ ret + chunks > h.actualChunksPerSegment && ret < h.actualChunksPerSegment)
+ freeList.clear(ret, h.actualChunksPerSegment);
+ ret = freeList.setNextNContinuousClearBits(0L, chunks);
+ if (ret == DirectBitSet.NOT_FOUND || ret + chunks > h.actualChunksPerSegment) {
+ if (ret != DirectBitSet.NOT_FOUND &&
+ ret + chunks > h.actualChunksPerSegment &&
+ ret < h.actualChunksPerSegment)
+ freeList.clear(ret, h.actualChunksPerSegment);
+ if (chunks == 1) {
+ throw new IllegalStateException(
+ "Segment is full, no free entries found");
+ } else {
+ throw new IllegalStateException(
+ "Segment is full or has no ranges of " + chunks
+ + " continuous free chunks"
+ );
+ }
+ }
+ updateNextPosToSearchFrom(ret, chunks);
+ } else {
+ // if bit at nextPosToSearchFrom is clear, it was skipped because
+ // more than 1 chunk was requested. Don't move nextPosToSearchFrom
+ // in this case. chunks == 1 clause is just a fast path.
+ if (chunks == 1 || freeList.isSet(nextPosToSearchFrom())) {
+ updateNextPosToSearchFrom(ret, chunks);
+ }
+ }
+ return ret;
+ }
+
+ public void free(long fromPos, int chunks) {
+ freeList.clear(fromPos, fromPos + chunks);
+ if (fromPos < nextPosToSearchFrom())
+ nextPosToSearchFrom(fromPos);
+ }
+
+ void updateNextPosToSearchFrom(long allocated, int chunks) {
+ long nextPosToSearchFrom = allocated + chunks;
+ if (nextPosToSearchFrom >= h.actualChunksPerSegment)
+ nextPosToSearchFrom = 0L;
+ nextPosToSearchFrom(nextPosToSearchFrom);
+ }
+
+
+ /////////////////////////////////////////////////
+ // Remove
+ @Override
+ public boolean remove() {
+ checkOnEachPublicOperation();
+ initRemoveDependencies();
+ try {
+ return remove0();
+ } finally {
+ closeRemove();
+ }
+ }
+
+ public boolean remove0() {
+ if (containsKey()) {
+ initEntrySizeInChunks();
+ upgradeToWriteLock();
+ hashLookupRemove();
+ free(pos, entrySizeInChunks);
+ entries(entries() - 1L);
+ searchState = DELETED;
+ return true;
+ } else {
+ return false;
+ }
+ }
+
+ public boolean forEachEntry;
+
+ public void hashLookupRemove() {
+ if (!forEachEntry)
+ hashLookup.remove();
+ }
+
+ public void initRemoveDependencies() {
+ initSegment();
+ initLocks();
+ updateLockIfNeeded();
+ }
+
+ public void closeRemove() {
+ }
+
+
+ public void clear() {
+ writeLock();
+ initSegment();
+ hashLookup.clear();
+ freeList.clear();
+ nextPosToSearchFrom(0L);
+ entries(0L);
+ }
+
+
+ /////////////////////////////////////////////////
+ // For bytes contexts
+ public static final Bytes DUMMY_BYTES = new ByteBufferBytes(ByteBuffer.allocate(0));
+ private final MultiStoreBytes keyCopy = new MultiStoreBytes();
+
+ public void initBytesKeyModel0() {
+ keyInterop = (KI) BytesBytesInterop.INSTANCE;
+ }
+
+ public void initBytesKey0(Bytes key) {
+ keySize = h.keySizeMarshaller.readSize(key);
+ key.limit(key.position() + keySize);
+ initBytesKey00(key);
+ metaKeyInterop = (MKI) DelegatingMetaBytesInterop.instance();
+ }
+
+ public void initBytesKey00(Bytes key) {
+ keyCopy.setBytesOffset(key, key.position());
+ keyCopy.limit(keySize);
+ this.key = (K) keyCopy;
+ }
+
+
+ /////////////////////////////////////////////////
+ // Iteration
+ public void initKeyFromPos() {
+ reuse(pos);
+ keySize = h.keySizeMarshaller.readSize(entry);
+ initKeyOffset0();
+ keyFound();
+ }
+
+ public K immutableKey() {
+ return key0(null);
+ }
+
+ K key0(K usingKey) {
+ initKeyReader();
+ entry.position(keyOffset);
+ key = keyReader.read(entry, keySize, usingKey);
+ return key;
+ }
+}
diff --git a/src/main/java/net/openhft/chronicle/map/HashSplitting.java b/src/main/java/net/openhft/chronicle/hash/impl/HashSplitting.java
similarity index 96%
rename from src/main/java/net/openhft/chronicle/map/HashSplitting.java
rename to src/main/java/net/openhft/chronicle/hash/impl/HashSplitting.java
index f32590d3e..fb5e4ee6a 100644
--- a/src/main/java/net/openhft/chronicle/map/HashSplitting.java
+++ b/src/main/java/net/openhft/chronicle/hash/impl/HashSplitting.java
@@ -14,13 +14,13 @@
* limitations under the License.
*/
-package net.openhft.chronicle.map;
+package net.openhft.chronicle.hash.impl;
import java.io.Serializable;
import static net.openhft.lang.Maths.isPowerOf2;
-interface HashSplitting extends Serializable {
+public interface HashSplitting extends Serializable {
int segmentIndex(long hash);
long segmentHash(long hash);
diff --git a/src/main/java/net/openhft/chronicle/map/SegmentHeader.java b/src/main/java/net/openhft/chronicle/hash/impl/SegmentHeader.java
similarity index 94%
rename from src/main/java/net/openhft/chronicle/map/SegmentHeader.java
rename to src/main/java/net/openhft/chronicle/hash/impl/SegmentHeader.java
index 1db4df704..7dc665cc5 100644
--- a/src/main/java/net/openhft/chronicle/map/SegmentHeader.java
+++ b/src/main/java/net/openhft/chronicle/hash/impl/SegmentHeader.java
@@ -14,7 +14,7 @@
* limitations under the License.
*/
-package net.openhft.chronicle.map;
+package net.openhft.chronicle.hash.impl;
import java.util.concurrent.TimeUnit;
@@ -22,6 +22,9 @@ public interface SegmentHeader {
long size(long address);
void size(long address, long size);
+ long deleted(long address);
+ void deleted(long address, long deleted);
+
long nextPosToSearchFrom(long address);
void nextPosToSearchFrom(long address, long nextPosToSearchFrom);
diff --git a/src/main/java/net/openhft/chronicle/hash/impl/VanillaChronicleHash.java b/src/main/java/net/openhft/chronicle/hash/impl/VanillaChronicleHash.java
new file mode 100644
index 000000000..ff8a2d2ec
--- /dev/null
+++ b/src/main/java/net/openhft/chronicle/hash/impl/VanillaChronicleHash.java
@@ -0,0 +1,335 @@
+/*
+ * 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.hash.impl;
+
+import net.openhft.chronicle.hash.ChronicleHash;
+import net.openhft.chronicle.hash.KeyContext;
+import net.openhft.chronicle.hash.impl.hashlookup.HashLookup;
+import net.openhft.chronicle.hash.serialization.BytesReader;
+import net.openhft.chronicle.hash.serialization.SizeMarshaller;
+import net.openhft.chronicle.hash.serialization.internal.MetaBytesInterop;
+import net.openhft.chronicle.hash.serialization.internal.MetaProvider;
+import net.openhft.chronicle.hash.serialization.internal.SerializationBuilder;
+import net.openhft.chronicle.hash.impl.util.BuildVersion;
+import net.openhft.lang.Jvm;
+import net.openhft.lang.io.Bytes;
+import net.openhft.lang.io.BytesStore;
+import net.openhft.lang.io.MappedStore;
+import net.openhft.lang.io.NativeBytes;
+import net.openhft.lang.io.serialization.JDKObjectSerializer;
+import net.openhft.lang.threadlocal.Provider;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.Serializable;
+import java.nio.channels.FileChannel;
+
+import static java.lang.Long.numberOfTrailingZeros;
+import static java.lang.Math.max;
+import static net.openhft.lang.MemoryUnit.*;
+
+public abstract class VanillaChronicleHash,
+ C extends KeyContext> implements ChronicleHash, Serializable {
+
+ private static final long serialVersionUID = 0L;
+
+ /////////////////////////////////////////////////
+ // Version
+ public final String dataFileVersion;
+
+ /////////////////////////////////////////////////
+ // Key Data model
+ public final Class kClass;
+ public final SizeMarshaller keySizeMarshaller;
+ public final BytesReader originalKeyReader;
+ public final KI originalKeyInterop;
+ public final MKI originalMetaKeyInterop;
+ public final MetaProvider metaKeyInteropProvider;
+
+ public transient Provider> keyReaderProvider;
+ transient Provider keyInteropProvider;
+
+ /////////////////////////////////////////////////
+ // Concurrency (number of segments), memory management and dependent fields
+ public final int actualSegments;
+ final HashSplitting hashSplitting;
+
+ public final long entriesPerSegment;
+
+ public final long chunkSize;
+ public final int maxChunksPerEntry;
+ public final long actualChunksPerSegment;
+
+ /////////////////////////////////////////////////
+ // Precomputed offsets and sizes for fast Context init
+ final int segmentHeaderSize;
+
+ final int segmentHashLookupValueBits;
+ final int segmentHashLookupKeyBits;
+ final int segmentHashLookupEntrySize;
+ final long segmentHashLookupCapacity;
+ final long segmentHashLookupInnerSize;
+ final long segmentHashLookupOuterSize;
+
+ final long segmentFreeListInnerSize;
+ final long segmentFreeListOuterSize;
+
+ final long segmentEntrySpaceInnerSize;
+ final int segmentEntrySpaceInnerOffset;
+ final long segmentEntrySpaceOuterSize;
+
+ final long segmentSize;
+
+ /////////////////////////////////////////////////
+ // Bytes Store (essentially, the base address) and serialization-dependent offsets
+ public transient BytesStore ms;
+ transient Bytes bytes;
+
+ public transient long headerSize;
+ transient long segmentHeadersOffset;
+ transient long segmentsOffset;
+
+ public VanillaChronicleHash(ChronicleHashBuilderImpl builder, boolean replicated) {
+ // Version
+ dataFileVersion = BuildVersion.version();
+
+ // Data model
+ SerializationBuilder keyBuilder = builder.keyBuilder();
+ kClass = keyBuilder.eClass;
+ keySizeMarshaller = keyBuilder.sizeMarshaller();
+ originalKeyReader = keyBuilder.reader();
+ originalKeyInterop = (KI) keyBuilder.interop();
+ originalMetaKeyInterop = (MKI) keyBuilder.metaInterop();
+ metaKeyInteropProvider = (MetaProvider) keyBuilder.metaInteropProvider();
+
+ actualSegments = builder.actualSegments(replicated);
+ hashSplitting = HashSplitting.Splitting.forSegments(actualSegments);
+
+ entriesPerSegment = builder.entriesPerSegment(replicated);
+
+ chunkSize = builder.chunkSize(replicated);
+ maxChunksPerEntry = builder.maxChunksPerEntry();
+ actualChunksPerSegment = builder.actualChunksPerSegment(replicated);
+
+ // Precomputed offsets and sizes for fast Context init
+ segmentHeaderSize = builder.segmentHeaderSize(replicated);
+
+ segmentHashLookupValueBits = HashLookup.valueBits(actualChunksPerSegment);
+ segmentHashLookupKeyBits =
+ HashLookup.keyBits(entriesPerSegment, segmentHashLookupValueBits);
+ segmentHashLookupEntrySize =
+ HashLookup.entrySize(segmentHashLookupKeyBits, segmentHashLookupValueBits);
+ segmentHashLookupCapacity = HashLookup.capacityFor(entriesPerSegment);
+ segmentHashLookupInnerSize = segmentHashLookupCapacity * segmentHashLookupEntrySize;
+ segmentHashLookupOuterSize = CACHE_LINES.align(segmentHashLookupInnerSize, BYTES);
+
+ segmentFreeListInnerSize = LONGS.align(
+ BYTES.alignAndConvert(actualChunksPerSegment, BITS), BYTES);
+ segmentFreeListOuterSize = CACHE_LINES.align(segmentFreeListInnerSize, BYTES);
+
+ segmentEntrySpaceInnerSize = chunkSize * actualChunksPerSegment;
+ segmentEntrySpaceInnerOffset = builder.segmentEntrySpaceInnerOffset(replicated);
+ segmentEntrySpaceOuterSize = CACHE_LINES.align(
+ segmentEntrySpaceInnerOffset + segmentEntrySpaceInnerSize, BYTES);
+
+ segmentSize = segmentSize();
+ }
+
+ private long segmentSize() {
+ long ss = segmentHashLookupOuterSize
+ + segmentFreeListOuterSize
+ + segmentEntrySpaceOuterSize;
+ if ((ss & 63L) != 0)
+ throw new AssertionError();
+ return breakL1CacheAssociativityContention(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 |= CACHE_LINES.toBytes(1L); // make segment size "odd" (in cache lines)
+ }
+ return segmentSize;
+ }
+
+ public void initTransients() {
+ ownInitTransients();
+ }
+
+ private void ownInitTransients() {
+ keyReaderProvider = Provider.of((Class) originalKeyReader.getClass());
+ keyInteropProvider = Provider.of((Class) originalKeyInterop.getClass());
+ }
+
+ public final void createMappedStoreAndSegments(BytesStore bytesStore) throws IOException {
+ this.ms = bytesStore;
+ bytes = ms.bytes();
+
+ onHeaderCreated();
+
+ segmentHeadersOffset = mapHeaderOuterSize();
+ long segmentHeadersSize = actualSegments * segmentHeaderSize;
+ segmentsOffset = segmentHeadersOffset + segmentHeadersSize;
+ }
+
+ public void warnOnWindows() {
+ if (!Jvm.isWindows())
+ return;
+ long offHeapMapSize = sizeInBytes();
+ long oneGb = GIGABYTES.toBytes(1L);
+ double offHeapMapSizeInGb = offHeapMapSize * 1.0 / oneGb;
+ if (offHeapMapSize > GIGABYTES.toBytes(4L)) {
+ System.out.printf(
+ "WARNING: On Windows, you probably cannot create a ChronicleMap\n" +
+ "of more than 4 GB. The configured map requires %.2f GB of " +
+ "off-heap memory.\n",
+ offHeapMapSizeInGb);
+ }
+ try {
+ long freePhysicalMemory = Jvm.freePhysicalMemoryOnWindowsInBytes();
+ if (offHeapMapSize > freePhysicalMemory * 0.9) {
+ double freePhysicalMemoryInGb = freePhysicalMemory * 1.0 / oneGb;
+ System.out.printf(
+ "WARNING: On Windows, you probably cannot create a ChronicleMap\n" +
+ "of more than 90%% of available free memory in the system.\n" +
+ "The configured map requires %.2f GB of off-heap memory.\n" +
+ "There is only %.2f GB of free physical memory in the system.\n",
+ offHeapMapSizeInGb, freePhysicalMemoryInGb);
+
+ }
+ } catch (IOException e) {
+ // ignore -- anyway we just warn the user
+ }
+ }
+
+ public final void createMappedStoreAndSegments(File file) throws IOException {
+ warnOnWindows();
+ createMappedStoreAndSegments(new MappedStore(file, FileChannel.MapMode.READ_WRITE,
+ sizeInBytes(), JDKObjectSerializer.INSTANCE));
+ }
+
+ private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException {
+ in.defaultReadObject();
+ ownInitTransients();
+ }
+
+ public void onHeaderCreated() {
+ }
+
+ /**
+ * @return the version of Chronicle Map that was used to create the current data file
+ */
+ public String persistedDataVersion() {
+ return dataFileVersion;
+ }
+
+ /**
+ * @return the version of chronicle map that is currently running
+ */
+ public String applicationVersion() {
+ return BuildVersion.version();
+ }
+
+ private long mapHeaderOuterSize() {
+ // Align segment headers on page boundary to minimize number of pages that
+ // segment headers span
+ long pageMask = NativeBytes.UNSAFE.pageSize() - 1L;
+ return (mapHeaderInnerSize() + pageMask) & ~pageMask;
+ }
+
+ public long mapHeaderInnerSize() {
+ return headerSize;
+ }
+
+ @Override
+ public File file() {
+ return ms.file();
+ }
+
+ public final long sizeInBytes() {
+ return mapHeaderOuterSize() + actualSegments * (segmentHeaderSize + segmentSize);
+ }
+
+ @Override
+ public void close() {
+ if (ms == null)
+ return;
+ bytes.release();
+ bytes = null;
+ ms.free();
+ ms = null;
+ }
+
+ public final void checkKey(Object key) {
+ if (!kClass.isInstance(key)) {
+ // key.getClass will cause NPE exactly as needed
+ throw new ClassCastException("Key must be a " + kClass.getName() +
+ " but was a " + key.getClass());
+ }
+ }
+
+ /**
+ * For testing
+ */
+ public final long[] segmentSizes() {
+ long[] sizes = new long[actualSegments];
+ for (int i = 0; i < actualSegments; i++) {
+ sizes[i] = BigSegmentHeader.INSTANCE.size(ms.address() + segmentHeaderOffset(i));
+ }
+ return sizes;
+ }
+
+ public final long segmentHeaderOffset(int segmentIndex) {
+ return segmentHeadersOffset + ((long) segmentIndex) * segmentHeaderSize;
+ }
+
+ public final long segmentOffset(int segmentIndex) {
+ return segmentsOffset + ((long) segmentIndex) * segmentSize;
+ }
+
+ public final int inChunks(long sizeInBytes) {
+ // TODO optimize for the case when chunkSize is power of 2, that is default (and often) now
+ if (sizeInBytes <= chunkSize)
+ return 1;
+ // int division is MUCH faster than long on Intel CPUs
+ sizeInBytes -= 1L;
+ if (sizeInBytes <= Integer.MAX_VALUE)
+ return (((int) sizeInBytes) / (int) chunkSize) + 1;
+ return (int) (sizeInBytes / chunkSize) + 1;
+ }
+
+ @Override
+ public final long longSize() {
+ long result = 0L;
+ for (int i = 0; i < actualSegments; i++) {
+ long segmentHeaderAddress = ms.address() + segmentHeaderOffset(i);
+ result += BigSegmentHeader.INSTANCE.size(segmentHeaderAddress) -
+ BigSegmentHeader.INSTANCE.deleted(segmentHeaderAddress);
+ }
+ return result;
+ }
+
+ public final int size() {
+ long size = longSize();
+ return size > Integer.MAX_VALUE ? Integer.MAX_VALUE : (int) size;
+ }
+}
diff --git a/src/main/java/net/openhft/chronicle/hash/impl/hashlookup/EntryConsumer.java b/src/main/java/net/openhft/chronicle/hash/impl/hashlookup/EntryConsumer.java
new file mode 100644
index 000000000..a7561e292
--- /dev/null
+++ b/src/main/java/net/openhft/chronicle/hash/impl/hashlookup/EntryConsumer.java
@@ -0,0 +1,21 @@
+/*
+ * 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.hash.impl.hashlookup;
+
+public interface EntryConsumer {
+ void accept(long key, long value);
+}
diff --git a/src/main/java/net/openhft/chronicle/map/HashLookup.java b/src/main/java/net/openhft/chronicle/hash/impl/hashlookup/HashLookup.java
similarity index 89%
rename from src/main/java/net/openhft/chronicle/map/HashLookup.java
rename to src/main/java/net/openhft/chronicle/hash/impl/hashlookup/HashLookup.java
index f599d6a38..d3587ec44 100644
--- a/src/main/java/net/openhft/chronicle/map/HashLookup.java
+++ b/src/main/java/net/openhft/chronicle/hash/impl/hashlookup/HashLookup.java
@@ -14,7 +14,7 @@
* limitations under the License.
*/
-package net.openhft.chronicle.map;
+package net.openhft.chronicle.hash.impl.hashlookup;
import net.openhft.lang.Maths;
@@ -22,17 +22,17 @@ import static net.openhft.lang.MemoryUnit.BITS;
import static net.openhft.lang.MemoryUnit.BYTES;
import static net.openhft.lang.io.NativeBytes.UNSAFE;
-class HashLookup {
+public class HashLookup {
// to fit 64 bits per slot.
- static final int MAX_SEGMENT_CHUNKS = 1 << 30;
- static final int MAX_SEGMENT_ENTRIES = 1 << 29;
+ public static final int MAX_SEGMENT_CHUNKS = 1 << 30;
+ public static final int MAX_SEGMENT_ENTRIES = 1 << 29;
- static int valueBits(long actualChunksPerSegment) {
+ public static int valueBits(long actualChunksPerSegment) {
return 64 - Long.numberOfLeadingZeros(actualChunksPerSegment - 1L);
}
- static int keyBits(long entriesPerSegment, int valueBits) {
+ public static int keyBits(long entriesPerSegment, int valueBits) {
// key hash cardinality is between 1.0 and 2.0 of key cardinality
int minKeyBits = 64 - Long.numberOfLeadingZeros(entriesPerSegment - 1L);
// This minimizes probability of hash collision => on search, additional cache line touch
@@ -44,11 +44,11 @@ class HashLookup {
return actualEntryBits - valueBits;
}
- static int entrySize(int keyBits, int valueBits) {
+ public static int entrySize(int keyBits, int valueBits) {
return (int) BYTES.alignAndConvert((long) (keyBits + valueBits), BITS);
}
- static long capacityFor(long entriesPerSegment) {
+ public static long capacityFor(long entriesPerSegment) {
if (entriesPerSegment < 0L)
throw new IllegalArgumentException("entriesPerSegment should be positive");
long capacity = Maths.nextPower2(entriesPerSegment, 64L);
@@ -83,7 +83,7 @@ class HashLookup {
private long searchStartPos = -1L;
private long searchPos = -1L;
- void reuse(long address, long capacity, int entrySize, int keyBits, int valueBits) {
+ public void reuse(long address, long capacity, int entrySize, int keyBits, int valueBits) {
this.address = address;
this.capacityMask = capacity - 1L;
@@ -267,20 +267,11 @@ class HashLookup {
@Override
public String toString() {
final StringBuilder sb = new StringBuilder("{");
- forEach(new EntryConsumer() {
- @Override
- public void accept(long key, long value) {
- sb.append(key).append('=').append(value).append(',');
- }
- });
+ forEach((key, value) -> sb.append(key).append('=').append(value).append(','));
sb.append('}');
return sb.toString();
}
- static interface EntryConsumer {
- void accept(long key, long value);
- }
-
public void forEach(EntryConsumer action) {
for (long pos = 0L; pos <= capacityMask2; pos += entrySize) {
long entry = readEntry(pos);
@@ -289,11 +280,7 @@ class HashLookup {
}
}
- static interface EntryPredicate {
- boolean remove(long hash, long pos);
- }
-
- public void forEachRemoving(EntryPredicate predicate) {
+ public void forEachRemoving(HashLookupIteration iteration) {
long pos = 0L;
while (!empty(readEntry(pos))) {
pos = step(pos);
@@ -303,11 +290,12 @@ class HashLookup {
pos = step(pos);
long entry = readEntry(pos);
if (!empty(entry)) {
- if (predicate.remove(key(entry), value(entry))) {
+ iteration.accept(key(entry), value(entry));
+ if (iteration.remove()) {
remove00(pos);
pos = stepBack(pos);
}
}
- } while (pos != startPos);
+ } while (pos != startPos && iteration.continueIteration());
}
}
diff --git a/src/main/java/net/openhft/chronicle/hash/function/Predicate.java b/src/main/java/net/openhft/chronicle/hash/impl/hashlookup/HashLookupIteration.java
similarity index 50%
rename from src/main/java/net/openhft/chronicle/hash/function/Predicate.java
rename to src/main/java/net/openhft/chronicle/hash/impl/hashlookup/HashLookupIteration.java
index 05fd9b54f..84b1564fa 100644
--- a/src/main/java/net/openhft/chronicle/hash/function/Predicate.java
+++ b/src/main/java/net/openhft/chronicle/hash/impl/hashlookup/HashLookupIteration.java
@@ -14,26 +14,12 @@
* limitations under the License.
*/
-package net.openhft.chronicle.hash.function;
+package net.openhft.chronicle.hash.impl.hashlookup;
+public interface HashLookupIteration {
+ void accept(long hash, long pos);
-/**
- * Represents a predicate (boolean-valued function) of one argument.
- *
- * This is a functional interface whose functional method is {@link #test(Object)}.
- *
- *
This is a copy of Java 8's {@code java.util.function.Predicate} interface.
- *
- * @param the type of the input to the predicate
- */
-public interface Predicate {
+ boolean remove();
- /**
- * Evaluates this predicate on the given argument.
- *
- * @param t the input argument
- * @return {@code true} if the input argument matches the predicate,
- * otherwise {@code false}
- */
- boolean test(T t);
+ boolean continueIteration();
}
diff --git a/src/main/java/net/openhft/chronicle/map/BuildVersion.java b/src/main/java/net/openhft/chronicle/hash/impl/util/BuildVersion.java
similarity index 95%
rename from src/main/java/net/openhft/chronicle/map/BuildVersion.java
rename to src/main/java/net/openhft/chronicle/hash/impl/util/BuildVersion.java
index 1912296a7..684afc81d 100644
--- a/src/main/java/net/openhft/chronicle/map/BuildVersion.java
+++ b/src/main/java/net/openhft/chronicle/hash/impl/util/BuildVersion.java
@@ -16,8 +16,9 @@
* limitations under the License.
*/
-package net.openhft.chronicle.map;
+package net.openhft.chronicle.hash.impl.util;
+import net.openhft.chronicle.map.ChronicleMapBuilder;
import shaded.org.apache.maven.model.Model;
import shaded.org.apache.maven.model.io.xpp3.MavenXpp3Reader;
@@ -26,7 +27,7 @@ import java.io.*;
/**
* gets the version of the current build
*/
-class BuildVersion {
+public final class BuildVersion {
private static String version = null;
@@ -106,6 +107,6 @@ class BuildVersion {
return null;
}
}
-
-
+
+ private BuildVersion() {}
}
\ No newline at end of file
diff --git a/src/main/java/net/openhft/chronicle/map/CloseablesManager.java b/src/main/java/net/openhft/chronicle/hash/impl/util/CloseablesManager.java
old mode 100755
new mode 100644
similarity index 95%
rename from src/main/java/net/openhft/chronicle/map/CloseablesManager.java
rename to src/main/java/net/openhft/chronicle/hash/impl/util/CloseablesManager.java
index ee8db28f9..a39161ca5
--- a/src/main/java/net/openhft/chronicle/map/CloseablesManager.java
+++ b/src/main/java/net/openhft/chronicle/hash/impl/util/CloseablesManager.java
@@ -16,7 +16,7 @@
* limitations under the License.
*/
-package net.openhft.chronicle.map;
+package net.openhft.chronicle.hash.impl.util;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -26,7 +26,7 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
-final class CloseablesManager implements Closeable {
+public final class CloseablesManager implements Closeable {
private static final Logger LOG = LoggerFactory.getLogger(CloseablesManager.class.getName());
private final List closeables = new ArrayList<>();
diff --git a/src/main/java/net/openhft/chronicle/map/Objects.java b/src/main/java/net/openhft/chronicle/hash/impl/util/Objects.java
similarity index 76%
rename from src/main/java/net/openhft/chronicle/map/Objects.java
rename to src/main/java/net/openhft/chronicle/hash/impl/util/Objects.java
index 6994c31dc..ffb233fdd 100644
--- a/src/main/java/net/openhft/chronicle/map/Objects.java
+++ b/src/main/java/net/openhft/chronicle/hash/impl/util/Objects.java
@@ -16,7 +16,7 @@
* limitations under the License.
*/
-package net.openhft.chronicle.map;
+package net.openhft.chronicle.hash.impl.util;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
@@ -26,23 +26,25 @@ import java.util.Arrays;
/**
* java.util.Objects since Java 7
*/
-final class Objects {
- static int hash(Object... values) {
+public final class Objects {
+ public static int hash(Object... values) {
return Arrays.hashCode(values);
}
- static boolean equal(@Nullable Object a, @Nullable Object b) {
+ public static boolean equal(@Nullable Object a, @Nullable Object b) {
return a != null ? a.equals(b) : b == null;
}
- static boolean builderEquals(@NotNull Object builder, @Nullable Object o) {
+ public static boolean builderEquals(@NotNull Object builder, @Nullable Object o) {
return builder == o ||
o != null && builder.getClass() == o.getClass() &&
builder.toString().equals(o.toString());
}
- static void requireNonNull(Object obj) {
+ public static void requireNonNull(Object obj) {
if (obj == null)
throw new NullPointerException();
}
+
+ private Objects() {}
}
diff --git a/src/main/java/net/openhft/chronicle/map/ReaderWithSize.java b/src/main/java/net/openhft/chronicle/hash/serialization/internal/ReaderWithSize.java
similarity index 90%
rename from src/main/java/net/openhft/chronicle/map/ReaderWithSize.java
rename to src/main/java/net/openhft/chronicle/hash/serialization/internal/ReaderWithSize.java
index 714efa832..7963d6f1e 100644
--- a/src/main/java/net/openhft/chronicle/map/ReaderWithSize.java
+++ b/src/main/java/net/openhft/chronicle/hash/serialization/internal/ReaderWithSize.java
@@ -14,8 +14,9 @@
* limitations under the License.
*/
-package net.openhft.chronicle.map;
+package net.openhft.chronicle.hash.serialization.internal;
+import net.openhft.chronicle.hash.serialization.internal.SerializationBuilder;
import net.openhft.chronicle.hash.serialization.BytesReader;
import net.openhft.chronicle.hash.serialization.SizeMarshaller;
import net.openhft.lang.io.Bytes;
@@ -23,12 +24,12 @@ import net.openhft.lang.threadlocal.Provider;
import net.openhft.lang.threadlocal.ThreadLocalCopies;
import org.jetbrains.annotations.Nullable;
-final class ReaderWithSize {
+public final class ReaderWithSize {
private final SizeMarshaller sizeMarshaller;
private final BytesReader originalReader;
private final Provider> readerProvider;
- ReaderWithSize(SerializationBuilder serializationBuilder) {
+ public ReaderWithSize(SerializationBuilder serializationBuilder) {
sizeMarshaller = serializationBuilder.sizeMarshaller();
originalReader = serializationBuilder.reader();
readerProvider = (Provider>) Provider.of(originalReader.getClass());
diff --git a/src/main/java/net/openhft/chronicle/map/SerializationBuilder.java b/src/main/java/net/openhft/chronicle/hash/serialization/internal/SerializationBuilder.java
similarity index 97%
rename from src/main/java/net/openhft/chronicle/map/SerializationBuilder.java
rename to src/main/java/net/openhft/chronicle/hash/serialization/internal/SerializationBuilder.java
index ca22cc032..e06c7975a 100644
--- a/src/main/java/net/openhft/chronicle/map/SerializationBuilder.java
+++ b/src/main/java/net/openhft/chronicle/hash/serialization/internal/SerializationBuilder.java
@@ -16,11 +16,9 @@
* limitations under the License.
*/
-package net.openhft.chronicle.map;
+package net.openhft.chronicle.hash.serialization.internal;
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.io.ByteBufferBytes;
import net.openhft.lang.io.Bytes;
import net.openhft.lang.io.serialization.BytesMarshallable;
@@ -45,9 +43,9 @@ import java.util.List;
import static net.openhft.chronicle.hash.serialization.internal.SizeMarshallers.constant;
import static net.openhft.chronicle.hash.serialization.internal.SizeMarshallers.stopBit;
-import static net.openhft.chronicle.map.Objects.hash;
+import static net.openhft.chronicle.hash.impl.util.Objects.hash;
-final class SerializationBuilder implements Cloneable, Serializable {
+public final class SerializationBuilder implements Cloneable, Serializable {
private static final long serialVersionUID = 0L;
private static final Bytes EMPTY_BYTES = new ByteBufferBytes(ByteBuffer.allocate(0));
@@ -71,12 +69,12 @@ final class SerializationBuilder implements Cloneable, Serializable {
return !knownJDKImmutableClasses.contains(c);
}
- enum Role {KEY, VALUE}
+ public enum Role {KEY, VALUE}
private enum CopyingInterop {FROM_MARSHALLER, FROM_WRITER}
private final Role role;
- final Class eClass;
+ public final Class eClass;
private boolean instancesAreMutable;
private SizeMarshaller sizeMarshaller = stopBit();
private BytesReader reader;
@@ -86,12 +84,12 @@ final class SerializationBuilder implements Cloneable, Serializable {
private MetaProvider metaInteropProvider;
private long maxSize;
- final boolean sizeIsStaticallyKnown;
+ public final boolean sizeIsStaticallyKnown;
boolean serializesGeneratedClasses = false;
@SuppressWarnings("unchecked")
- SerializationBuilder(Class eClass, Role role) {
+ public SerializationBuilder(Class eClass, Role role) {
this.role = role;
this.eClass = eClass;
configureByDefault(eClass, role);
@@ -191,7 +189,7 @@ final class SerializationBuilder implements Cloneable, Serializable {
in.defaultReadObject();
}
- boolean possibleOffHeapReferences() {
+ public boolean possibleOffHeapReferences() {
if (reader instanceof CharSequenceReader)
return false;
// exclude some known classes, most notably boxed primitive types
@@ -323,7 +321,7 @@ final class SerializationBuilder implements Cloneable, Serializable {
return sizeMarshaller;
}
- boolean constantSizeMarshaller() {
+ public boolean constantSizeMarshaller() {
try {
return sizeMarshaller().sizeEncodingSize(1L) == 0;
} catch (Exception e) {
@@ -334,11 +332,11 @@ final class SerializationBuilder implements Cloneable, Serializable {
}
}
- boolean constantSizeEncodingSizeMarshaller() {
+ public boolean constantSizeEncodingSizeMarshaller() {
return sizeMarshaller().minSizeEncodingSize() == sizeMarshaller().maxSizeEncodingSize();
}
- long pseudoReadConstantSize() {
+ public long pseudoReadConstantSize() {
return sizeMarshaller().readSize(EMPTY_BYTES);
}
diff --git a/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java b/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java
index 7bd73d280..885139a28 100644
--- a/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java
+++ b/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java
@@ -19,11 +19,11 @@
package net.openhft.chronicle.map;
import net.openhft.chronicle.hash.replication.ThrottlingConfig;
+import net.openhft.chronicle.hash.impl.util.CloseablesManager;
import net.openhft.chronicle.map.ReplicatedChronicleMap.BytesReplicatedContextFactory;
import net.openhft.lang.io.ByteBufferBytes;
import net.openhft.lang.io.Bytes;
import net.openhft.lang.thread.NamedThreadFactory;
-import net.openhft.lang.threadlocal.ThreadLocalCopies;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
import org.slf4j.Logger;
diff --git a/src/main/java/net/openhft/chronicle/map/AbstractChronicleMap.java b/src/main/java/net/openhft/chronicle/map/AbstractChronicleMap.java
index 87c68c5da..81063c02c 100644
--- a/src/main/java/net/openhft/chronicle/map/AbstractChronicleMap.java
+++ b/src/main/java/net/openhft/chronicle/map/AbstractChronicleMap.java
@@ -17,40 +17,37 @@
package net.openhft.chronicle.map;
import net.openhft.chronicle.hash.KeyContext;
-import net.openhft.chronicle.hash.function.Consumer;
-import net.openhft.chronicle.hash.function.Function;
-import net.openhft.chronicle.hash.function.Predicate;
-import net.openhft.chronicle.map.ReplicatedChronicleMap.ReplicatedContext;
+import net.openhft.chronicle.hash.function.SerializableFunction;
+import net.openhft.chronicle.hash.impl.hashlookup.HashLookupIteration;
import org.jetbrains.annotations.NotNull;
import java.io.File;
import java.io.IOException;
import java.io.Serializable;
import java.util.*;
+import java.util.function.BiConsumer;
+import java.util.function.Consumer;
+import java.util.function.Predicate;
import static java.util.Collections.emptyList;
-import static net.openhft.chronicle.map.Objects.requireNonNull;
-import static net.openhft.chronicle.map.VanillaContext.SearchState.DELETED;
+import static net.openhft.chronicle.hash.impl.HashContext.SearchState.DELETED;
+import static net.openhft.chronicle.hash.impl.util.Objects.requireNonNull;
-abstract class AbstractChronicleMap extends AbstractMap
- implements ChronicleMap, Serializable {
- private static final long serialVersionUID = 0L;
-
- transient Set> entrySet;
+public interface AbstractChronicleMap extends ChronicleMap, Serializable {
@Override
- public abstract VanillaContext context(K key);
+ VanillaContext context(K key);
@Override
@SuppressWarnings("unchecked")
- public final boolean containsKey(Object key) {
+ default boolean containsKey(Object key) {
try (MapKeyContext c = context((K) key)) {
return c.containsKey();
}
}
@Override
- public final V put(K key, V value) {
+ default V put(K key, V value) {
try (MapKeyContext c = context(key)) {
// We cannot read the previous value under read lock, because then we will need
// to release the read lock -> acquire write lock, the value might be updated in
@@ -63,12 +60,12 @@ abstract class AbstractChronicleMap extends AbstractMap
}
}
- V prevValueOnPut(MapKeyContext context) {
+ default V prevValueOnPut(MapKeyContext context) {
return context.getUsing(null);
}
@Override
- public final V putIfAbsent(K key, V value) {
+ default V putIfAbsent(K key, V value) {
checkValue(value);
try (MapKeyContext c = context(key)) {
// putIfAbsent() shouldn't actually put most of the time,
@@ -94,23 +91,23 @@ abstract class AbstractChronicleMap extends AbstractMap
}
}
- abstract void checkValue(V value);
+ void checkValue(V value);
@Override
@SuppressWarnings("unchecked")
- public final V get(Object key) {
+ default V get(Object key) {
return getUsing((K) key, null);
}
@Override
- public final V getUsing(K key, V usingValue) {
+ default V getUsing(K key, V usingValue) {
try (MapKeyContext c = context(key)) {
return c.getUsing(usingValue);
}
}
@Override
- public final V acquireUsing(K key, V usingValue) {
+ default V acquireUsing(K key, V usingValue) {
try (VanillaContext c = context(key)) {
// acquireUsing() should just read an existing value most of the time,
// so try to check if the key is already present under read lock first:
@@ -135,17 +132,17 @@ abstract class AbstractChronicleMap extends AbstractMap
}
}
- void upgradeReadToUpdateLockWithUnlockingIfNeeded(KeyContext c) {
+ default void upgradeReadToUpdateLockWithUnlockingIfNeeded(KeyContext c) {
if (!c.updateLock().tryLock()) {
c.readLock().unlock();
c.updateLock().lock();
}
}
- abstract void putDefaultValue(VanillaContext context);
+ void putDefaultValue(VanillaContext context);
@Override
- public R getMapped(K key, @NotNull Function super V, R> function) {
+ default R getMapped(K key, @NotNull SerializableFunction super V, R> function) {
requireNonNull(function);
try (MapKeyContext c = context(key)) {
return c.containsKey() ? function.apply(c.get()) : null;
@@ -153,7 +150,7 @@ abstract class AbstractChronicleMap extends AbstractMap
}
@Override
- public V putMapped(K key, @NotNull UnaryOperator unaryOperator) {
+ default V putMapped(K key, @NotNull UnaryOperator unaryOperator) {
requireNonNull(unaryOperator);
try (MapKeyContext c = context(key)) {
// putMapped() should find a value, update & put most of the time,
@@ -171,18 +168,22 @@ abstract class AbstractChronicleMap extends AbstractMap
}
@Override
- public synchronized void getAll(File toFile) throws IOException {
- JsonSerializer.getAll(toFile, this, emptyList());
+ default void getAll(File toFile) throws IOException {
+ synchronized (this) {
+ JsonSerializer.getAll(toFile, this, emptyList());
+ }
}
@Override
- public synchronized void putAll(File fromFile) throws IOException {
- JsonSerializer.putAll(fromFile, this, emptyList());
+ default void putAll(File fromFile) throws IOException {
+ synchronized (this) {
+ JsonSerializer.putAll(fromFile, this, emptyList());
+ }
}
@Override
@SuppressWarnings("unchecked")
- public final V remove(Object key) {
+ default V remove(Object key) {
try (MapKeyContext c = context((K) key)) {
// We cannot read the previous value under read lock, because then we will need
// to release the read lock -> acquire write lock, the value might be updated in
@@ -195,13 +196,13 @@ abstract class AbstractChronicleMap extends AbstractMap
}
}
- V prevValueOnRemove(MapKeyContext context) {
+ default V prevValueOnRemove(MapKeyContext context) {
return context.getUsing(null);
}
@Override
@SuppressWarnings("unchecked")
- public final boolean remove(Object key, Object value) {
+ default boolean remove(Object key, Object value) {
if (value == null)
return false; // CHM compatibility; General ChronicleMap policy is to throw NPE
checkValue((V) value);
@@ -215,7 +216,7 @@ abstract class AbstractChronicleMap extends AbstractMap
}
@Override
- public final V replace(K key, V value) {
+ default V replace(K key, V value) {
checkValue(value);
try (MapKeyContext c = context(key)) {
// replace(key, value) should find the key & put the value most of the time,
@@ -234,7 +235,7 @@ abstract class AbstractChronicleMap extends AbstractMap
}
@Override
- public final boolean replace(K key, V oldValue, V newValue) {
+ default boolean replace(K key, V oldValue, V newValue) {
checkValue(oldValue);
checkValue(newValue);
try (MapKeyContext c = context(key)) {
@@ -246,180 +247,174 @@ abstract class AbstractChronicleMap extends AbstractMap
}
}
- abstract int actualSegments();
+ int actualSegments();
- abstract VanillaContext rawContext();
+ VanillaContext mapContext();
- abstract VanillaContext mapContext();
+ @Override
+ default boolean containsValue(Object value) {
+ return !forEachEntryWhile(c -> !c.valueEqualTo((V) value));
+ }
+
+ @Override
+ default boolean isEmpty() {
+ return size() == 0;
+ }
+
+ @Override
+ default void forEach(BiConsumer super K, ? super V> action) {
+ forEachEntry(c -> action.accept(c.key(), c.get()));
+ }
+
+ @Override
+ default void putAll(Map extends K, ? extends V> m) {
+ m.forEach(this::put);
+ }
@NotNull
@Override
- public final Set> entrySet() {
- return (entrySet != null) ? entrySet : (entrySet = newEntrySet());
+ default Collection values() {
+ // todo cache view object
+ return new AbstractCollection() {
+ @Override
+ public Iterator iterator() {
+ return new Iterator() {
+ private Iterator> i = entrySet().iterator();
+
+ @Override
+ public boolean hasNext() {
+ return i.hasNext();
+ }
+
+ @Override
+ public V next() {
+ return i.next().getValue();
+ }
+
+ @Override
+ public void remove() {
+ i.remove();
+ }
+ };
+ }
+
+ @Override
+ public int size() {
+ return AbstractChronicleMap.this.size();
+ }
+
+ @Override
+ public boolean isEmpty() {
+ return AbstractChronicleMap.this.isEmpty();
+ }
+
+ @Override
+ public void clear() {
+ AbstractChronicleMap.this.clear();
+ }
+
+ @Override
+ public boolean contains(Object v) {
+ return AbstractChronicleMap.this.containsValue(v);
+ }
+
+ @Override
+ public void forEach(java.util.function.Consumer super V> action) {
+ AbstractChronicleMap.this.forEachEntry(c -> action.accept(c.get()));
+ }
+ };
}
- Set> newEntrySet() {
- return new EntrySet();
+ @NotNull
+ @Override
+ default Set keySet() {
+ // todo cache view object
+ return new AbstractSet() {
+ @Override
+ public Iterator iterator() {
+ return new ChronicleMapIterator.OfKeys<>(AbstractChronicleMap.this);
+ }
+
+ @Override
+ public int size() {
+ return AbstractChronicleMap.this.size();
+ }
+
+ public boolean isEmpty() {
+ return AbstractChronicleMap.this.isEmpty();
+ }
+
+ public void clear() {
+ AbstractChronicleMap.this.clear();
+ }
+
+ public boolean contains(Object k) {
+ return AbstractChronicleMap.this.containsKey(k);
+ }
+
+ @Override
+ public void forEach(java.util.function.Consumer super K> action) {
+ AbstractChronicleMap.this.forEachEntry(c -> action.accept(c.key()));
+ }
+ };
}
+ default boolean mapEquals(Object o) {
+ if (o == this)
+ return true;
- class EntryIterator implements Iterator>, HashLookup.EntryConsumer {
- /**
- * Very inefficient (esp. if segments are large), but CORRECT implementation
- */
- private final Thread ownerThread = Thread.currentThread();
- private int segmentIndex = actualSegments() - 1;
- private final Queue> entryBuffer = new ArrayDeque<>();
- private VanillaContext context;
- private Entry returned;
+ if (!(o instanceof Map))
+ return false;
+ Map,?> m = (Map,?>) o;
+ if ((m instanceof ChronicleMap ? ((ChronicleMap) m).longSize() : m.size()) != longSize())
+ return false;
- private void checkSingleThreaded() {
- if (ownerThread != Thread.currentThread()) {
- throw new IllegalStateException(
- "Iterator should be accessed only from a single thread");
- }
- }
-
- private void fillEntryBuffer() {
- if (!entryBuffer.isEmpty())
- return;
- while (true) {
- if (segmentIndex < 0)
- return;
- try (VanillaContext c = mapContext()) {
- context = c;
- c.segmentIndex = segmentIndex;
- segmentIndex--;
- if (c.size() == 0)
- continue;
- c.updateLock().lock();
- c.initSegment();
- c.hashLookup.forEach(this);
- return;
- } finally {
- context = null;
- }
- }
- }
-
- @Override
- public void accept(long hash, long pos) {
- context.pos = pos;
- context.initKeyFromPos();
- try {
- if (!context.containsKey()) // for replicated map
- return;
- K key = context.immutableKey();
- V value = context.getUsing(null);
- entryBuffer.add(new WriteThroughEntry(key, value));
- } finally {
- context.closeKeySearch();
- }
- }
-
- @Override
- public boolean hasNext() {
- checkSingleThreaded();
- fillEntryBuffer();
- return !entryBuffer.isEmpty();
- }
-
- @Override
- public Entry next() {
- checkSingleThreaded();
- fillEntryBuffer();
- Entry e;
- if ((e = entryBuffer.poll()) == null)
- throw new NoSuchElementException();
- return returned = e;
- }
-
- @Override
- public void remove() {
- checkSingleThreaded();
- if (returned == null)
- throw new IllegalStateException();
- AbstractChronicleMap.this.remove(returned.getKey(), returned.getValue());
- returned = null;
+ try {
+ return forEachEntryWhile(c -> c.get().equals(m.get(c.key())));
+ } catch (ClassCastException unused) {
+ return false;
+ } catch (NullPointerException unused) {
+ return false;
}
}
- class WriteThroughEntry extends AbstractMap.SimpleEntry {
-
- public WriteThroughEntry(K key, V value) {
- super(key, value);
- }
-
- @Override
- public V setValue(V value) {
- try (MapKeyContext c = AbstractChronicleMap.this.context(getKey())) {
- c.put(value);
- }
- return super.setValue(value);
- }
+ default int mapHashCode() {
+ int[] h = new int[1];
+ forEach((k, v) -> h[0] += k.hashCode() ^ v.hashCode());
+ return h[0];
}
- class EntrySet extends AbstractSet> {
- @NotNull
- public Iterator> iterator() {
- return new EntryIterator();
- }
-
- public final boolean contains(Object o) {
- if (!(o instanceof Map.Entry))
- return false;
- Map.Entry, ?> e = (Map.Entry, ?>) o;
- try {
- V v = AbstractChronicleMap.this.get(e.getKey());
- return v != null && v.equals(e.getValue());
- } catch (ClassCastException ex) {
- return false;
- } catch (NullPointerException ex) {
- return false;
- }
- }
-
- public final boolean remove(Object o) {
- if (!(o instanceof Map.Entry))
- return false;
- Map.Entry, ?> e = (Map.Entry, ?>) o;
- try {
- Object key = e.getKey();
- Object value = e.getValue();
- return AbstractChronicleMap.this.remove(key, value);
- } catch (ClassCastException ex) {
- return false;
- } catch (NullPointerException ex) {
- return false;
- }
- }
-
- public final int size() {
- return AbstractChronicleMap.this.size();
- }
-
- public final boolean isEmpty() {
- return AbstractChronicleMap.this.isEmpty();
- }
-
- public final void clear() {
- AbstractChronicleMap.this.clear();
- }
+ default String mapToString() {
+ if (isEmpty())
+ return "{}";
+ StringBuilder sb = new StringBuilder();
+ sb.append('{');
+ forEach((k, v) -> sb
+ .append(k != AbstractChronicleMap.this ? k : "(this Map)")
+ .append('=')
+ .append(v != AbstractChronicleMap.this ? v : "(this Map)")
+ .append(',').append(' '));
+ if (sb.length() > 2)
+ sb.setLength(sb.length() - 2);
+ sb.append('}');
+ return sb.toString();
}
+ default Set> newEntrySet() {
+ return new ChronicleMapEntrySet<>(this);
+ }
+
+
@Override
- public void forEachEntry(final Consumer super MapKeyContext> action) {
- forEachEntryWhile(new Predicate>() {
- @Override
- public boolean test(MapKeyContext c) {
- action.accept(c);
- return true;
- }
+ default void forEachEntry(final Consumer super MapKeyContext> action) {
+ forEachEntryWhile(c -> {
+ action.accept(c);
+ return true;
});
}
@Override
- public boolean forEachEntryWhile(final Predicate super MapKeyContext> predicate) {
+ default boolean forEachEntryWhile(final Predicate super MapKeyContext> predicate) {
boolean interrupt = false;
iteration:
try (VanillaContext c = mapContext()) {
@@ -448,9 +443,10 @@ abstract class AbstractChronicleMap extends AbstractMap
}
- private static class ForEachWhilePredicate implements HashLookup.EntryPredicate {
+ static class ForEachWhilePredicate implements HashLookupIteration {
private final VanillaContext c;
private final Predicate super MapKeyContext> predicate;
+ boolean shouldRemove = false;
boolean shouldBreak = false;
public ForEachWhilePredicate(VanillaContext c,
@@ -461,12 +457,14 @@ abstract class AbstractChronicleMap extends AbstractMap
}
@Override
- public boolean remove(long hash, long pos) {
+ public void accept(long hash, long pos) {
c.pos = pos;
c.initKeyFromPos();
try {
- if (!c.containsKey()) // for replicated map
- return false;
+ if (!c.containsKey()) { // for replicated map
+ shouldRemove = false;
+ return;
+ }
shouldBreak = !predicate.test(c);
c.closeKey0();
// release all exclusive locks: possibly if context.remove() is performed
@@ -478,10 +476,20 @@ abstract class AbstractChronicleMap extends AbstractMap
throw new IllegalStateException("Shouldn't release update lock " +
"inside forEachEntry() callback");
}
- return c.state == DELETED && !(c instanceof ReplicatedContext);
+ shouldRemove = c.searchState0() == DELETED;
} finally {
c.closeKeySearch();
}
}
+
+ @Override
+ public boolean remove() {
+ return shouldRemove;
+ }
+
+ @Override
+ public boolean continueIteration() {
+ return !shouldBreak;
+ }
}
}
diff --git a/src/main/java/net/openhft/chronicle/map/BytesChronicleMap.java b/src/main/java/net/openhft/chronicle/map/BytesChronicleMap.java
index c2afca4ed..f552148d7 100644
--- a/src/main/java/net/openhft/chronicle/map/BytesChronicleMap.java
+++ b/src/main/java/net/openhft/chronicle/map/BytesChronicleMap.java
@@ -16,12 +16,14 @@
package net.openhft.chronicle.map;
+import net.openhft.chronicle.hash.impl.util.Objects;
import net.openhft.lang.io.Bytes;
import org.jetbrains.annotations.NotNull;
import java.io.File;
+import java.util.Set;
-public class BytesChronicleMap extends AbstractChronicleMap {
+public class BytesChronicleMap implements AbstractChronicleMap {
private final VanillaChronicleMap, ?, ?, ?, ?, ?> delegate;
TcpReplicator.TcpSocketChannelEntryWriter output;
@@ -31,22 +33,17 @@ public class BytesChronicleMap extends AbstractChronicleMap {
}
@Override
- void putDefaultValue(VanillaContext context) {
+ public void putDefaultValue(VanillaContext context) {
delegate.putDefaultValue(context);
}
@Override
- int actualSegments() {
+ public int actualSegments() {
return delegate.actualSegments();
}
@Override
- VanillaContext rawContext() {
- return delegate.rawBytesContext();
- }
-
- @Override
- VanillaContext mapContext() {
+ public VanillaContext mapContext() {
VanillaContext context = delegate.bytesMapContext();
context.output = output;
return context;
@@ -71,7 +68,7 @@ public class BytesChronicleMap extends AbstractChronicleMap {
}
@Override
- void checkValue(Bytes value) {
+ public void checkValue(Bytes value) {
Objects.requireNonNull(value);
}
@@ -122,4 +119,20 @@ public class BytesChronicleMap extends AbstractChronicleMap {
entries.position(entryPosition);
}
}
+
+ @Override
+ public int size() {
+ return delegate.size();
+ }
+
+ @Override
+ public void clear() {
+ delegate.clear();
+ }
+
+ @NotNull
+ @Override
+ public Set> entrySet() {
+ throw new UnsupportedOperationException();
+ }
}
diff --git a/src/main/java/net/openhft/chronicle/map/ChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ChronicleMap.java
index fec6c58c1..eb252567c 100755
--- a/src/main/java/net/openhft/chronicle/map/ChronicleMap.java
+++ b/src/main/java/net/openhft/chronicle/map/ChronicleMap.java
@@ -19,9 +19,7 @@
package net.openhft.chronicle.map;
import net.openhft.chronicle.hash.ChronicleHash;
-import net.openhft.chronicle.hash.KeyContext;
-import net.openhft.chronicle.hash.function.Consumer;
-import net.openhft.chronicle.hash.function.Function;
+import net.openhft.chronicle.hash.function.SerializableFunction;
import net.openhft.chronicle.hash.serialization.BytesReader;
import net.openhft.lang.io.Bytes;
import net.openhft.lang.io.serialization.BytesMarshaller;
@@ -151,7 +149,7 @@ public interface ChronicleMap
* @param return type.
* @return the result of the function, or null if there is no entry for the key.
*/
- R getMapped(K key, @NotNull Function super V, R> function);
+ R getMapped(K key, @NotNull SerializableFunction super V, R> function);
/**
* Apply a unaryOperator to the value for a key and return a result. A write lock is assumed.
diff --git a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java
index 689f60fac..ab7bbe7ef 100755
--- a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java
+++ b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java
@@ -20,6 +20,9 @@ package net.openhft.chronicle.map;
import net.openhft.chronicle.hash.ChronicleHashBuilder;
import net.openhft.chronicle.hash.ChronicleHashInstanceBuilder;
+import net.openhft.chronicle.hash.impl.ChronicleHashBuilderImpl;
+import net.openhft.chronicle.hash.impl.hashlookup.HashLookup;
+import net.openhft.chronicle.hash.serialization.internal.SerializationBuilder;
import net.openhft.chronicle.hash.replication.*;
import net.openhft.chronicle.hash.serialization.*;
import net.openhft.chronicle.hash.serialization.internal.MetaBytesInterop;
@@ -53,7 +56,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import static java.lang.Math.round;
-import static net.openhft.chronicle.map.Objects.builderEquals;
+import static net.openhft.chronicle.hash.impl.util.Objects.builderEquals;
import static net.openhft.lang.model.DataValueGenerator.firstPrimitiveFieldType;
/**
@@ -99,7 +102,7 @@ import static net.openhft.lang.model.DataValueGenerator.firstPrimitiveFieldType;
* @see ChronicleSetBuilder
*/
public final class ChronicleMapBuilder implements
- ChronicleHashBuilder, ChronicleMapBuilder>,
+ ChronicleHashBuilderImpl, ChronicleMapBuilder>,
MapBuilder>, Serializable {
static final byte UDP_REPLICATION_MODIFICATION_ITERATOR_ID = (byte) 128;
@@ -392,6 +395,11 @@ public final class ChronicleMapBuilder implements
return this;
}
+ @Override
+ public SerializationBuilder keyBuilder() {
+ return keyBuilder;
+ }
+
static class EntrySizeInfo {
final double averageEntrySize;
final int worstAlignment;
@@ -451,7 +459,8 @@ public final class ChronicleMapBuilder implements
* This is needed, if chunkSize = constant entry size is not aligned, for entry alignment to be
* always the same, we should _misalign_ the first chunk.
*/
- int segmentEntrySpaceInnerOffset(boolean replicated) {
+ @Override
+ public int segmentEntrySpaceInnerOffset(boolean replicated) {
if (!constantlySizedEntries())
return 0;
return (int) (constantValueSize() % valueAlignment().alignment());
@@ -505,7 +514,8 @@ public final class ChronicleMapBuilder implements
return greatestCommonDivisor(b, a % b);
}
- long chunkSize(boolean replicated) {
+ @Override
+ public long chunkSize(boolean replicated) {
if (actualChunkSize > 0)
return actualChunkSize;
double averageEntrySize = entrySizeInfo(replicated).averageEntrySize;
@@ -546,7 +556,8 @@ public final class ChronicleMapBuilder implements
return this;
}
- int maxChunksPerEntry() {
+ @Override
+ public int maxChunksPerEntry() {
if (maxChunksPerEntry > 0)
return maxChunksPerEntry;
if (constantlySizedEntries()) {
@@ -627,7 +638,8 @@ public final class ChronicleMapBuilder implements
return this;
}
- long entriesPerSegment(boolean replicated) {
+ @Override
+ public long entriesPerSegment(boolean replicated) {
long entriesPerSegment;
if (this.entriesPerSegment > 0L) {
entriesPerSegment = this.entriesPerSegment;
@@ -680,7 +692,8 @@ public final class ChronicleMapBuilder implements
}
}
- long actualChunksPerSegment(boolean replicated) {
+ @Override
+ public long actualChunksPerSegment(boolean replicated) {
if (actualChunksPerSegment > 0)
return actualChunksPerSegment;
return chunksPerSegment(entriesPerSegment(replicated), replicated);
@@ -769,7 +782,8 @@ public final class ChronicleMapBuilder implements
return this;
}
- int actualSegments(boolean replicated) {
+ @Override
+ public int actualSegments(boolean replicated) {
if (actualSegments > 0)
return actualSegments;
if (entriesPerSegment > 0) {
@@ -857,7 +871,8 @@ public final class ChronicleMapBuilder implements
return segments <= maxSegments ? segments : -segments;
}
- int segmentHeaderSize(boolean replicated) {
+ @Override
+ public int segmentHeaderSize(boolean replicated) {
int segments = actualSegments(replicated);
// reduce false sharing unless we have a lot of segments.
return segments <= 16 * 1024 ? 64 : 32;
@@ -1389,7 +1404,7 @@ public final class ChronicleMapBuilder implements
V, Object, MetaBytesInterop>(this, replication);
} else {
return new VanillaChronicleMap,
- V, Object, MetaBytesInterop>(this);
+ V, Object, MetaBytesInterop>(this, false);
}
}
diff --git a/src/main/java/net/openhft/chronicle/map/ChronicleMapEntrySet.java b/src/main/java/net/openhft/chronicle/map/ChronicleMapEntrySet.java
new file mode 100644
index 000000000..4d4da572f
--- /dev/null
+++ b/src/main/java/net/openhft/chronicle/map/ChronicleMapEntrySet.java
@@ -0,0 +1,78 @@
+/*
+ * 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.jetbrains.annotations.NotNull;
+
+import java.util.AbstractSet;
+import java.util.Iterator;
+import java.util.Map;
+
+class ChronicleMapEntrySet extends AbstractSet> {
+
+ private final AbstractChronicleMap map;
+
+ public ChronicleMapEntrySet(AbstractChronicleMap map) {
+ this.map = map;
+ }
+
+ @NotNull
+ public Iterator> iterator() {
+ return new ChronicleMapIterator.OfEntries<>(map);
+ }
+
+ public final boolean contains(Object o) {
+ if (!(o instanceof Map.Entry))
+ return false;
+ Map.Entry, ?> e = (Map.Entry, ?>) o;
+ try {
+ V v = map.get(e.getKey());
+ return v != null && v.equals(e.getValue());
+ } catch (ClassCastException ex) {
+ return false;
+ } catch (NullPointerException ex) {
+ return false;
+ }
+ }
+
+ public final boolean remove(Object o) {
+ if (!(o instanceof Map.Entry))
+ return false;
+ Map.Entry, ?> e = (Map.Entry, ?>) o;
+ try {
+ Object key = e.getKey();
+ Object value = e.getValue();
+ return map.remove(key, value);
+ } catch (ClassCastException ex) {
+ return false;
+ } catch (NullPointerException ex) {
+ return false;
+ }
+ }
+
+ public final int size() {
+ return map.size();
+ }
+
+ public final boolean isEmpty() {
+ return map.isEmpty();
+ }
+
+ public final void clear() {
+ map.clear();
+ }
+}
diff --git a/src/main/java/net/openhft/chronicle/map/ChronicleMapIterator.java b/src/main/java/net/openhft/chronicle/map/ChronicleMapIterator.java
new file mode 100644
index 000000000..4539e9ee1
--- /dev/null
+++ b/src/main/java/net/openhft/chronicle/map/ChronicleMapIterator.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.chronicle.hash.impl.hashlookup.EntryConsumer;
+
+import java.util.*;
+
+/**
+ * Very inefficient (esp. if segments are large), but CORRECT implementation
+ */
+abstract class ChronicleMapIterator implements Iterator, EntryConsumer {
+
+ final AbstractChronicleMap map;
+ private final Thread ownerThread = Thread.currentThread();
+ private final Queue entryBuffer = new ArrayDeque<>();
+ private int segmentIndex;
+ VanillaContext context;
+ E returned;
+
+ ChronicleMapIterator(AbstractChronicleMap map) {
+ this.map = map;
+ segmentIndex = map.actualSegments() - 1;
+ }
+
+ private void checkSingleThreaded() {
+ if (ownerThread != Thread.currentThread()) {
+ throw new IllegalStateException(
+ "Iterator should be accessed only from a single thread");
+ }
+ }
+
+ private void fillEntryBuffer() {
+ if (!entryBuffer.isEmpty())
+ return;
+ while (true) {
+ if (segmentIndex < 0)
+ return;
+ try (VanillaContext c = map.mapContext()) {
+ context = c;
+ c.segmentIndex = segmentIndex;
+ segmentIndex--;
+ if (c.size() == 0)
+ continue;
+ c.updateLock().lock();
+ c.initSegment();
+ c.hashLookup.forEach(this);
+ return;
+ } finally {
+ context = null;
+ }
+ }
+ }
+
+ @Override
+ public void accept(long hash, long pos) {
+ context.pos = pos;
+ context.initKeyFromPos();
+ try {
+ if (!context.containsKey()) // for replicated map
+ return;
+ entryBuffer.add(read());
+ } finally {
+ context.closeKeySearch();
+ }
+ }
+
+ abstract E read();
+
+ @Override
+ public boolean hasNext() {
+ checkSingleThreaded();
+ fillEntryBuffer();
+ return !entryBuffer.isEmpty();
+ }
+
+ @Override
+ public E next() {
+ checkSingleThreaded();
+ fillEntryBuffer();
+ E e;
+ if ((e = entryBuffer.poll()) == null)
+ throw new NoSuchElementException();
+ return returned = e;
+ }
+
+ @Override
+ public void remove() {
+ checkSingleThreaded();
+ if (returned == null)
+ throw new IllegalStateException();
+ removeReturned();
+ returned = null;
+ }
+
+ abstract void removeReturned();
+
+ static class OfEntries extends ChronicleMapIterator> {
+
+ OfEntries(AbstractChronicleMap map) {
+ super(map);
+ }
+
+ @Override
+ Map.Entry read() {
+ K key = context.immutableKey();
+ V value = context.getUsing(null);
+ return new WriteThroughEntry<>(map, key, value);
+ }
+
+ @Override
+ void removeReturned() {
+ map.remove(returned.getKey(), returned.getValue());
+ }
+ }
+
+ static class OfKeys extends ChronicleMapIterator {
+
+ OfKeys(AbstractChronicleMap map) {
+ super(map);
+ }
+
+ @Override
+ K read() {
+ return context.immutableKey();
+ }
+
+ @Override
+ void removeReturned() {
+ map.remove(returned);
+ }
+ }
+}
diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java
index a2538010c..9b13bb51a 100644
--- a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java
+++ b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java
@@ -18,7 +18,10 @@
package net.openhft.chronicle.map;
+import net.openhft.chronicle.hash.KeyContext;
import net.openhft.chronicle.hash.hashing.LongHashFunction;
+import net.openhft.chronicle.hash.impl.ContextFactory;
+import net.openhft.chronicle.hash.impl.HashContext;
import net.openhft.chronicle.hash.replication.AbstractReplication;
import net.openhft.chronicle.hash.replication.LateUpdateException;
import net.openhft.chronicle.hash.replication.TimeProvider;
@@ -26,7 +29,6 @@ import net.openhft.chronicle.hash.serialization.BytesReader;
import net.openhft.chronicle.hash.serialization.internal.BytesBytesInterop;
import net.openhft.chronicle.hash.serialization.internal.DelegatingMetaBytesInterop;
import net.openhft.chronicle.hash.serialization.internal.MetaBytesInterop;
-import net.openhft.chronicle.map.VanillaContext.ContextFactory;
import net.openhft.lang.Maths;
import net.openhft.lang.collection.ATSDirectBitSet;
import net.openhft.lang.collection.SingleThreadedDirectBitSet;
@@ -38,13 +40,11 @@ import org.slf4j.LoggerFactory;
import java.io.Closeable;
import java.io.IOException;
-import java.util.Iterator;
+import java.io.ObjectInputStream;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.atomic.AtomicReferenceArray;
-import static net.openhft.chronicle.map.VanillaContext.SearchState.DELETED;
-import static net.openhft.chronicle.map.VanillaContext.SearchState.PRESENT;
import static net.openhft.lang.MemoryUnit.*;
import static net.openhft.lang.collection.DirectBitSet.NOT_FOUND;
@@ -109,7 +109,7 @@ final class ReplicatedChronicleMap builder,
AbstractReplication replication)
throws IOException {
- super(builder);
+ super(builder, true);
this.timeProvider = builder.timeProvider();
this.localIdentifier = replication.identifier();
@@ -125,13 +125,22 @@ final class ReplicatedChronicleMap(127 + RESERVED_MOD_ITER);
closeables = new CopyOnWriteArraySet<>();
}
+ private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException {
+ in.defaultReadObject();
+ ownInitTransients();
+ }
+
long modIterBitSetSizeInBytes() {
long bytes = BITS.toBytes(bitsPerSegmentInModIterBitSet() * actualSegments);
return CACHE_LINES.align(bytes, BYTES);
@@ -144,7 +153,7 @@ final class ReplicatedChronicleMap> iterator = entrySet().iterator(); iterator.hasNext(); ) {
- iterator.next();
- iterator.remove();
- }
+ forEachEntry(KeyContext::remove);
}
void addCloseable(Closeable closeable) {
@@ -280,16 +284,16 @@ final class ReplicatedChronicleMap m() {
- return (ReplicatedChronicleMap) m;
+ ReplicatedChronicleMap rm() {
+ return (ReplicatedChronicleMap) m();
}
@Override
- void totalCheckClosed() {
+ public void totalCheckClosed() {
super.totalCheckClosed();
assert !replicationStateInit() : "replication state not closed";
assert !replicationUpdateInit() : "replication update not closed";
@@ -302,19 +306,28 @@ final class ReplicatedChronicleMap {
@Override
@@ -494,14 +498,13 @@ final class ReplicatedChronicleMap) new PseudoMetaBytesInterop();
@Override
- boolean remove0() {
+ public boolean remove0() {
if (containsKey()) {
if (!shouldIgnore()) {
writeReplicationBytes();
writeDeleted();
updateChange();
- size(size() - 1L);
- state = DELETED;
+ deleted(deleted() + 1);
return true;
} else {
return false;
@@ -515,12 +518,11 @@ final class ReplicatedChronicleMap newIdentifier;
}
if (shouldIgnore && !remoteUpdate()) {
- assert newIdentifier == m().identifier();
+ assert newIdentifier == rm().identifier();
throw new LateUpdateException(identifier, timestamp, newIdentifier, newTimestamp);
}
return shouldIgnore;
@@ -562,17 +564,14 @@ final class ReplicatedChronicleMap 0L;
- if (timestamp >= dirtyFromTimeStamp && (!bootstrapOnlyLocalEntries ||
- identifier == m().identifier()))
- modIter.raiseChange(segmentIndex, pos);
- }
+ hashLookup.forEach((hash, pos) -> {
+ ReplicatedContext.this.pos = pos;
+ initKeyFromPos();
+ keyFound();
+ assert timestamp > 0L;
+ if (timestamp >= dirtyFromTimeStamp && (!bootstrapOnlyLocalEntries ||
+ identifier == rm().identifier()))
+ modIter.raiseChange(segmentIndex, pos);
});
} finally {
readLock().unlock();
@@ -584,8 +583,7 @@ final class ReplicatedChronicleMap acquireContext(K key) {
ReplicatedAcquireContext c =
VanillaContext.get(ReplicatedAcquireContextFactory.INSTANCE);
- c.initMap(this);
+ c.initHash(this);
c.initKey(key);
return c;
}
@@ -694,7 +692,7 @@ final class ReplicatedChronicleMap> newEntrySet() {
- return new EntrySet();
- }
-
@Override
public K key(@NotNull Bytes entry, K usingKey) {
final long start = entry.position();
diff --git a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java
index ff319671b..6d21da108 100644
--- a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java
+++ b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java
@@ -20,10 +20,12 @@ package net.openhft.chronicle.map;
import com.sun.jdi.connect.spi.ClosedConnectionException;
import net.openhft.chronicle.hash.RemoteCallTimeoutException;
-import net.openhft.chronicle.hash.function.Consumer;
-import net.openhft.chronicle.hash.function.Function;
-import net.openhft.chronicle.hash.function.Predicate;
+import net.openhft.chronicle.hash.function.SerializableFunction;
+import net.openhft.chronicle.hash.serialization.internal.ReaderWithSize;
+import net.openhft.chronicle.hash.serialization.internal.SerializationBuilder;
import net.openhft.chronicle.hash.serialization.BytesReader;
+import net.openhft.chronicle.hash.impl.util.BuildVersion;
+import net.openhft.chronicle.hash.impl.util.CloseablesManager;
import net.openhft.lang.io.ByteBufferBytes;
import net.openhft.lang.io.Bytes;
import net.openhft.lang.io.NativeBytes;
@@ -44,6 +46,8 @@ import java.nio.channels.SocketChannel;
import java.util.*;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.Consumer;
+import java.util.function.Predicate;
import static java.nio.ByteBuffer.allocateDirect;
import static java.util.Collections.emptyList;
@@ -610,7 +614,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon
}
@Nullable
- public R getMapped(@Nullable K key, @NotNull Function super V, R> function) {
+ public R getMapped(@Nullable K key, @NotNull SerializableFunction super V, R> function) {
if (key == null)
throw keyNotNullNPE();
return fetchObject(MAP_FOR_KEY, key, function);
diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java
index 02cef950d..9b792c1d9 100644
--- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java
+++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java
@@ -18,11 +18,14 @@
package net.openhft.chronicle.map;
-import net.openhft.chronicle.hash.function.Function;
+import net.openhft.chronicle.hash.function.SerializableFunction;
+import net.openhft.chronicle.hash.serialization.internal.ReaderWithSize;
+import net.openhft.chronicle.hash.serialization.internal.SerializationBuilder;
import net.openhft.chronicle.hash.replication.RemoteNodeValidator;
import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig;
import net.openhft.chronicle.hash.replication.ThrottlingConfig;
import net.openhft.chronicle.hash.serialization.BytesReader;
+import net.openhft.chronicle.hash.impl.util.BuildVersion;
import net.openhft.lang.io.AbstractBytes;
import net.openhft.lang.io.ByteBufferBytes;
import net.openhft.lang.io.Bytes;
@@ -47,7 +50,7 @@ import static java.nio.channels.SelectionKey.*;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static net.openhft.chronicle.map.AbstractChannelReplicator.SIZE_OF_SIZE;
import static net.openhft.chronicle.map.AbstractChannelReplicator.SIZE_OF_TRANSACTION_ID;
-import static net.openhft.chronicle.map.BuildVersion.version;
+import static net.openhft.chronicle.hash.impl.util.BuildVersion.version;
import static net.openhft.chronicle.map.StatelessChronicleMap.EventId.HEARTBEAT;
interface Work {
@@ -1639,7 +1642,7 @@ class StatelessServerConnector {
@NotNull TcpReplicator.TcpSocketChannelEntryWriter writer,
long sizeLocation) {
final K key = keyReaderWithSize.read(reader, null, null);
- final Function function = (Function) reader.readObject();
+ final SerializableFunction function = (SerializableFunction) reader.readObject();
try {
Object result = map.getMapped(key, function);
writeObject(writer, result);
diff --git a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java
index 4a2ce5c0b..5b95af4fe 100755
--- a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java
+++ b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java
@@ -19,18 +19,13 @@
package net.openhft.chronicle.map;
import net.openhft.chronicle.hash.KeyContext;
+import net.openhft.chronicle.hash.impl.ContextFactory;
+import net.openhft.chronicle.hash.impl.HashContext;
+import net.openhft.chronicle.hash.impl.VanillaChronicleHash;
import net.openhft.chronicle.hash.serialization.BytesReader;
import net.openhft.chronicle.hash.serialization.SizeMarshaller;
-import net.openhft.chronicle.hash.serialization.internal.BytesBytesInterop;
-import net.openhft.chronicle.hash.serialization.internal.DelegatingMetaBytesInterop;
-import net.openhft.chronicle.hash.serialization.internal.MetaBytesInterop;
-import net.openhft.chronicle.hash.serialization.internal.MetaProvider;
-import net.openhft.chronicle.map.VanillaContext.ContextFactory;
-import net.openhft.lang.Jvm;
+import net.openhft.chronicle.hash.serialization.internal.*;
import net.openhft.lang.io.Bytes;
-import net.openhft.lang.io.BytesStore;
-import net.openhft.lang.io.MappedStore;
-import net.openhft.lang.io.serialization.JDKObjectSerializer;
import net.openhft.lang.model.DataValueClasses;
import net.openhft.lang.threadlocal.Provider;
import net.openhft.lang.threadlocal.ThreadLocalCopies;
@@ -38,37 +33,24 @@ import org.jetbrains.annotations.NotNull;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.io.File;
import java.io.IOException;
import java.io.ObjectInputStream;
-import java.nio.channels.FileChannel;
+import java.util.Map;
+import java.util.Set;
-import static java.lang.Long.numberOfTrailingZeros;
-import static java.lang.Math.max;
-import static net.openhft.chronicle.map.ChronicleMapBuilder.RUNTIME_PAGE_SIZE;
import static net.openhft.chronicle.map.ChronicleMapBuilder.greatestCommonDivisor;
-import static net.openhft.lang.MemoryUnit.*;
class VanillaChronicleMap,
- V, VI, MVI extends MetaBytesInterop> extends AbstractChronicleMap {
+ V, VI, MVI extends MetaBytesInterop>
+ extends VanillaChronicleHash>
+ implements AbstractChronicleMap {
private static final Logger LOG = LoggerFactory.getLogger(VanillaChronicleMap.class);
private static final long serialVersionUID = 3L;
/////////////////////////////////////////////////
- // Version
- final String dataFileVersion;
-
- /////////////////////////////////////////////////
- // Data model
- final Class kClass;
- final SizeMarshaller keySizeMarshaller;
- final BytesReader originalKeyReader;
- final KI originalKeyInterop;
- final MKI originalMetaKeyInterop;
- final MetaProvider metaKeyInteropProvider;
-
+ // Value Data model
final Class vClass;
final Class nativeValueClass;
final SizeMarshaller valueSizeMarshaller;
@@ -81,9 +63,6 @@ class VanillaChronicleMap,
final boolean constantlySizedEntry;
- transient Provider> keyReaderProvider;
- transient Provider keyInteropProvider;
-
transient Provider> valueReaderProvider;
transient Provider valueInteropProvider;
@@ -97,66 +76,16 @@ class VanillaChronicleMap