diff --git a/pom.xml b/pom.xml
index 5bb26408b..b41477b4e 100644
--- a/pom.xml
+++ b/pom.xml
@@ -47,7 +47,7 @@
net.openhftlang
- 6.4.22
+ 6.4.23-SNAPSHOT
diff --git a/src/main/java/net/openhft/chronicle/hash/ChronicleHash.java b/src/main/java/net/openhft/chronicle/hash/ChronicleHash.java
index b5a7a31fa..9025fb5da 100644
--- a/src/main/java/net/openhft/chronicle/hash/ChronicleHash.java
+++ b/src/main/java/net/openhft/chronicle/hash/ChronicleHash.java
@@ -28,7 +28,7 @@ import java.io.File;
* This interface defines common {@link ChronicleMap} and {@link ChronicleSet}, related to off-heap
* memory management and file-mapping. Not usable by itself.
*/
-public interface ChronicleHash extends Closeable {
+public interface ChronicleHash extends Closeable {
/**
* Returns the file this hash container mapped to, i. e. when it is created by
* {@link ChronicleHashBuilder#create()} call, or {@code null} if it is purely in-memory,
@@ -40,6 +40,15 @@ public interface ChronicleHash extends Closeable {
*/
File file();
+ long longSize();
+
+ KeyContext context(K key);
+
+ /**
+ * @return the class of {@code }
+ */
+ Class keyClass();
+
/**
* Releases the off-heap memory, used by this hash container and resources, used by replication,
* if any. However, if hash container (hence off-heap memory, used by it) is mapped to the file
diff --git a/src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilder.java b/src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilder.java
index 1d997ea2e..08b8849a4 100755
--- a/src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilder.java
+++ b/src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilder.java
@@ -37,7 +37,6 @@ import org.jetbrains.annotations.NotNull;
import java.io.Externalizable;
import java.io.File;
import java.io.IOException;
-import java.net.InetSocketAddress;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
@@ -61,13 +60,13 @@ import java.util.concurrent.TimeUnit;
* adjustments, etc.
*
* @param the type of keys in hash containers, created by this builder
- * @param the container type, created by this builder, i. e. {@link ChronicleMap} or {@link
+ * @param the container type, created by this builder, i. e. {@link ChronicleMap} or {@link
* ChronicleSet}
* @param the concrete builder type, i. e. {@link ChronicleMapBuilder}
* or {@link ChronicleSetBuilder}
*/
-public interface ChronicleHashBuilder> extends Cloneable {
+public interface ChronicleHashBuilder,
+ B extends ChronicleHashBuilder> extends Cloneable {
/**
* Clones this builder. Useful for configuration persisting, because {@code
@@ -257,28 +256,6 @@ public interface ChronicleHashBuilderDefault lock time out is 2 seconds.
- *
- * @param lockTimeOut new lock timeout for segments of containers created by this builder, in
- * the given time units
- * @param unit time unit of the given lock timeout
- * @return this builder back
- */
- B lockTimeOut(long lockTimeOut, TimeUnit unit);
-
- B errorListener(ChronicleHashErrorListener errorListener);
-
- B metaDataBytes(int metaDataBytes);
-
/**
* Configures a time provider, used by hash containers, created by this builder, for needs of
* replication consensus protocol (conflicting data updates resolution).
@@ -358,19 +335,19 @@ public interface ChronicleHashBuilder keyWriter, @NotNull BytesReader keyReader);
+ B keyMarshallers(@NotNull BytesWriter super K> keyWriter, @NotNull BytesReader keyReader);
/**
* Configures the marshaller used to serialize actual key sizes to off-heap memory in hash
* containers, created by this builder.
*
- *
Default key size marshaller is so-called {@linkplain SizeMarshallers#stopBit() stop bit
- * encoding marshalling}. If {@linkplain #constantKeySizeBySample(Object) constant key size} is
- * configured, or defaulted if the key type is always constant and {@code ChronicleHashBuilder}
- * implementation knows about it, this configuration takes no effect, because a special {@link
- * SizeMarshaller} implementation, which doesn't actually do any marshalling, and just returns
- * the known constant size on {@link SizeMarshaller#readSize(Bytes)} calls, is used instead of
- * any {@code SizeMarshaller} configured using this method.
+ *
Default key size marshaller is so-called "stop bit encoding" marshalling. If {@linkplain
+ * #constantKeySizeBySample(Object) constant key size} is configured, or defaulted if the key
+ * type is always constant and {@code ChronicleHashBuilder} implementation knows about it, this
+ * configuration takes no effect, because a special {@link SizeMarshaller} implementation, which
+ * doesn't actually do any marshalling, and just returns the known constant size on {@link
+ * SizeMarshaller#readSize(Bytes)} calls, is used instead of any {@code SizeMarshaller}
+ * configured using this method.
*
* @param keySizeMarshaller the new marshaller, used to serialize actual key sizes to off-heap
* memory
@@ -398,7 +375,7 @@ public interface ChronicleHashBuilder keyDeserializationFactory);
+ B keyDeserializationFactory(@NotNull ObjectFactory extends K> keyDeserializationFactory);
/**
* Specifies that key objects, queried with the hash containers, created by this builder, are
@@ -447,7 +424,7 @@ public interface ChronicleHashBuilder instance();
+ ChronicleHashInstanceBuilder instance();
/**
* Creates a new hash container, storing it's data in off-heap memory, not mapped to any file.
@@ -461,7 +438,7 @@ public interface ChronicleHashBuilder 1L << 32) {
- LoggerFactory.getLogger(getClass()).warn("Grabbing lock held by processId: {}, threadId: {}",
- (threadId >>> 33), (threadId & 0xFFFFFFL));
- } else {
- LoggerFactory.getLogger(getClass()).warn("Grabbing lock held by threadId: {}", threadId);
- }
- }
-
- @Override
- public void errorOnUnlock(IllegalMonitorStateException e) {
- LoggerFactory.getLogger(getClass()).warn("Failed to unlock as expected", e);
- }
- }
-
- private static enum ThrowingErrorListener implements ChronicleHashErrorListener {
- INSTANCE;
-
- @Override
- public void onLockTimeout(long threadId) throws IllegalStateException {
- throw new IllegalStateException("Unable to acquire lock held by threadId: " + threadId);
- }
-
- @Override
- public void errorOnUnlock(IllegalMonitorStateException e) {
- throw e;
- }
- }
-}
diff --git a/src/main/java/net/openhft/chronicle/hash/ChronicleHashInstanceBuilder.java b/src/main/java/net/openhft/chronicle/hash/ChronicleHashInstanceBuilder.java
index ee6c86575..1ec13b337 100644
--- a/src/main/java/net/openhft/chronicle/hash/ChronicleHashInstanceBuilder.java
+++ b/src/main/java/net/openhft/chronicle/hash/ChronicleHashInstanceBuilder.java
@@ -23,11 +23,12 @@ import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig;
import java.io.File;
import java.io.IOException;
-public interface ChronicleHashInstanceBuilder {
+public interface ChronicleHashInstanceBuilder {
- ChronicleHashInstanceBuilder replicated(byte identifier, TcpTransportAndNetworkConfig tcpTransportAndNetwork);
+ ChronicleHashInstanceBuilder replicated(
+ byte identifier, TcpTransportAndNetworkConfig tcpTransportAndNetwork);
- ChronicleHashInstanceBuilder replicated(SingleChronicleHashReplication replication);
+ ChronicleHashInstanceBuilder replicated(SingleChronicleHashReplication replication);
/**
* Configures replication of the hash containers, created by this builder, via so called
@@ -49,11 +50,11 @@ public interface ChronicleHashInstanceBuilder {
* @return this builder object back
* @see #replicated(SingleChronicleHashReplication)
*/
- ChronicleHashInstanceBuilder replicatedViaChannel(ReplicationChannel channel);
+ ChronicleHashInstanceBuilder replicatedViaChannel(ReplicationChannel channel);
- ChronicleHashInstanceBuilder persistedTo(File file);
+ ChronicleHashInstanceBuilder persistedTo(File file);
- ChronicleHashInstanceBuilder name(String name);
+ ChronicleHashInstanceBuilder name(String name);
- C create() throws IOException;
+ H create() throws IOException;
}
diff --git a/src/main/java/net/openhft/chronicle/hash/ChronicleHashStatelessClientBuilder.java b/src/main/java/net/openhft/chronicle/hash/ChronicleHashStatelessClientBuilder.java
index 153c36ac1..3519889aa 100644
--- a/src/main/java/net/openhft/chronicle/hash/ChronicleHashStatelessClientBuilder.java
+++ b/src/main/java/net/openhft/chronicle/hash/ChronicleHashStatelessClientBuilder.java
@@ -39,12 +39,12 @@ import java.util.concurrent.TimeUnit;
* {@code ChronicleSet}
*/
public interface ChronicleHashStatelessClientBuilder<
- C extends ChronicleHashStatelessClientBuilder, H extends ChronicleHash> {
- C timeout(long timeout, TimeUnit units);
+ B extends ChronicleHashStatelessClientBuilder, H extends ChronicleHash> {
+ B timeout(long timeout, TimeUnit units);
- C name(String name);
+ B name(String name);
- C tcpBufferSize(int tcpBufferSize);
+ B tcpBufferSize(int tcpBufferSize);
H create() throws IOException;
}
diff --git a/src/main/java/net/openhft/chronicle/hash/KeyContext.java b/src/main/java/net/openhft/chronicle/hash/KeyContext.java
new file mode 100644
index 000000000..e93246161
--- /dev/null
+++ b/src/main/java/net/openhft/chronicle/hash/KeyContext.java
@@ -0,0 +1,103 @@
+/*
+ * 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;
+
+import net.openhft.chronicle.hash.locks.InterProcessReadWriteUpdateLock;
+import net.openhft.lang.io.Bytes;
+import org.jetbrains.annotations.NotNull;
+
+import java.util.ConcurrentModificationException;
+
+/**
+ * Common considerations:
+ *
+ *
All method throw {@link ConcurrentModificationException} immediately, if called not from
+ * the thread, in which the context was obtained. {@code KeyContext} is strictly
+ * for single-thread access. For single try-with-resources block, actually, because:
+ *
All methods throw {@code IllegalStateException} immediately, if called after
+ * {@link #close()} called on this context and returned.
+ *
+ */
+public interface KeyContext extends AutoCloseable, InterProcessReadWriteUpdateLock {
+
+ /**
+ * Returns the entry bytes, if the key is present in the hash, otherwise throws
+ * {@code IllegalStateException}. Might acquire {@link #readLock} before searching for
+ * the entry, if the context is not locked yet.
+ *
+ * @return the entry bytes
+ * @throws IllegalStateException if the hash doesn't {@link #containsKey}
+ */
+ @NotNull
+ Bytes entry();
+
+ /**
+ * Returns the offset of the key bytes within the {@link #entry}, if the key is present
+ * in the hash, otherwise throws {@code IllegalStateException}. Might acquire {@link #readLock}
+ * before searching for the entry, if the context is not locked yet.
+ *
+ * @return the offset of the key bytes within the {@link #entry}
+ * @throws IllegalStateException if the hash doesn't {@link #containsKey}
+ */
+ long keyOffset();
+
+ /**
+ * Returns the size of the key bytes representation. Bytes from {@link #keyOffset} to
+ * {@code keyOffset + keySize} are the key bytes in the {@link #entry}.
+ *
+ * @return the size of the key bytes representation
+ */
+ long keySize();
+
+ /**
+ * Checks if the key is present in the hash. Might acquire {@link #readLock} before searching
+ * for the key, if the context is not locked yet.
+ *
+ * @return {@code true} if the key is present in the hash, {@code false} otherwise
+ * @throws IllegalStateException if at least {@link #readLock} is not held
+ */
+ boolean containsKey();
+
+ /**
+ * Removes the key from the hash. Might acquire {@link #updateLock} or {@link #writeLock},
+ * if the context is not locked yet. Might upgrade from update to write lock, if update lock
+ * is already held.
+ *
+ * @return {@code true} if the key was actually removed (hash was updated),
+ * {@code false} if the key is already absent in the hash, or the remove operation was rejected
+ * by the hash for some reason
+ * @throws IllegalStateException if the context is locked for read, but not for update, and
+ * update or write lock is actually required for {@code remove()} in this {@code KeyContext}
+ * implementation
+ */
+ boolean remove();
+
+ /**
+ * Closes the context. Among other things, it releases all locks, acquired within the context,
+ * that means you shouldn't bother with unlocking and try-finally pattern, if the whole
+ * context operation if performed within try-with-resources block:
{@code
+ * try (KeyContext c = map.context(key)) {
+ * c.writeLock().lock();
+ * if (random.nextBoolean())
+ * c.remove();
+ * // Note, there is no c.writeLock().unlock(), and it's OK,
+ * // because all locks are released in close()
+ * }}
+ */
+ @Override
+ void close();
+}
diff --git a/src/main/java/net/openhft/chronicle/hash/ChronicleHashErrorListener.java b/src/main/java/net/openhft/chronicle/hash/Segment.java
similarity index 66%
rename from src/main/java/net/openhft/chronicle/hash/ChronicleHashErrorListener.java
rename to src/main/java/net/openhft/chronicle/hash/Segment.java
index 903e553c3..b13e2619e 100644
--- a/src/main/java/net/openhft/chronicle/hash/ChronicleHashErrorListener.java
+++ b/src/main/java/net/openhft/chronicle/hash/Segment.java
@@ -1,7 +1,5 @@
/*
- * Copyright 2014 Higher Frequency Trading
- *
- * http://www.higherfrequencytrading.com
+ * 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.
@@ -18,10 +16,12 @@
package net.openhft.chronicle.hash;
-import java.io.Serializable;
+import net.openhft.chronicle.hash.locks.InterProcessReadWriteUpdateLock;
+import org.jetbrains.annotations.NotNull;
-public interface ChronicleHashErrorListener extends Serializable {
- void onLockTimeout(long threadId) throws IllegalStateException;
+public interface Segment {
+ long index();
- void errorOnUnlock(IllegalMonitorStateException e);
+ @NotNull
+ InterProcessReadWriteUpdateLock lock();
}
diff --git a/src/main/java/net/openhft/chronicle/map/ZeroOutValueBytes.java b/src/main/java/net/openhft/chronicle/hash/locks/IllegalInterProcessLockStateException.java
similarity index 61%
rename from src/main/java/net/openhft/chronicle/map/ZeroOutValueBytes.java
rename to src/main/java/net/openhft/chronicle/hash/locks/IllegalInterProcessLockStateException.java
index a6e2ce845..ad51d4fa0 100644
--- a/src/main/java/net/openhft/chronicle/map/ZeroOutValueBytes.java
+++ b/src/main/java/net/openhft/chronicle/hash/locks/IllegalInterProcessLockStateException.java
@@ -14,22 +14,11 @@
* limitations under the License.
*/
-package net.openhft.chronicle.map;
+package net.openhft.chronicle.hash.locks;
-import net.openhft.lang.io.Bytes;
-
-final class ZeroOutValueBytes implements PrepareValueBytes {
+public class IllegalInterProcessLockStateException extends IllegalMonitorStateException {
private static final long serialVersionUID = 0L;
- private final long valueSize;
-
- ZeroOutValueBytes(long valueSize) {
- this.valueSize = valueSize;
- }
-
- @Override
- public void prepare(Bytes bytes, K key) {
- long pos = bytes.position();
- bytes.zeroOut(pos, pos + valueSize);
- bytes.skip(valueSize);
+ public IllegalInterProcessLockStateException(String s) {
+ super(s);
}
}
diff --git a/src/main/java/net/openhft/chronicle/hash/locks/InterProcessLock.java b/src/main/java/net/openhft/chronicle/hash/locks/InterProcessLock.java
new file mode 100644
index 000000000..e4f4284af
--- /dev/null
+++ b/src/main/java/net/openhft/chronicle/hash/locks/InterProcessLock.java
@@ -0,0 +1,50 @@
+/*
+ * 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.locks;
+
+import org.jetbrains.annotations.NotNull;
+
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.Lock;
+
+/**
+ * - Inter-process,
+ * - unfair,
+ * - busy-waiting,
+ * - reentrant,
+ * - single-thread (paradox? :) Once a lock object obtained, it shouldn't be stored in a field and
+ * accessed from multiple threads, instead of that, lock objects should be obtained in each thread
+ * separately, using the same call chain. This is because since the lock is inter-process, it
+ * anyway keep the syncronization state in shared memory, but restricting on-heap "view" of
+ * shared lock is beneficial form perf POV, e. g., fields of the on-heap lock object shouldn't
+ * be volatile.
+ * - doesn't support conditions.
+ */
+public interface InterProcessLock extends Lock {
+
+ boolean isHeldByCurrentThread();
+
+ /**
+ * Conditions are not supported by inter-process locks.
+ *
+ * @return nothing, always throws an exception
+ * @throws UnsupportedOperationException always
+ */
+ @NotNull
+ @Override
+ Condition newCondition();
+}
diff --git a/src/main/java/net/openhft/chronicle/map/ReadValue.java b/src/main/java/net/openhft/chronicle/hash/locks/InterProcessReadWriteUpdateLock.java
similarity index 63%
rename from src/main/java/net/openhft/chronicle/map/ReadValue.java
rename to src/main/java/net/openhft/chronicle/hash/locks/InterProcessReadWriteUpdateLock.java
index 2b90e13d7..a20cd964e 100644
--- a/src/main/java/net/openhft/chronicle/map/ReadValue.java
+++ b/src/main/java/net/openhft/chronicle/hash/locks/InterProcessReadWriteUpdateLock.java
@@ -14,19 +14,23 @@
* limitations under the License.
*/
-package net.openhft.chronicle.map;
+package net.openhft.chronicle.hash.locks;
-import net.openhft.lang.io.Bytes;
-import net.openhft.lang.threadlocal.ThreadLocalCopies;
import org.jetbrains.annotations.NotNull;
-/**
- * @param result type, map value type or Bytes
- */
-interface ReadValue {
- /**
- * Updates the entry position by valueSize.
- */
- RV readValue(@NotNull ThreadLocalCopies copies, Bytes entry, RV usingValue, long valueSize);
- RV readNull();
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReadWriteLock;
+
+public interface InterProcessReadWriteUpdateLock extends ReadWriteLock {
+
+ @NotNull
+ @Override
+ InterProcessLock readLock();
+
+ @NotNull
+ InterProcessLock updateLock();
+
+ @NotNull
+ @Override
+ InterProcessLock writeLock();
}
diff --git a/src/main/java/net/openhft/chronicle/map/ReadContext.java b/src/main/java/net/openhft/chronicle/hash/locks/LockingStrategy.java
similarity index 65%
rename from src/main/java/net/openhft/chronicle/map/ReadContext.java
rename to src/main/java/net/openhft/chronicle/hash/locks/LockingStrategy.java
index 3792464c6..6f6a336ad 100644
--- a/src/main/java/net/openhft/chronicle/map/ReadContext.java
+++ b/src/main/java/net/openhft/chronicle/hash/locks/LockingStrategy.java
@@ -1,7 +1,5 @@
/*
- * Copyright 2014 Higher Frequency Trading
- *
- * http://www.higherfrequencytrading.com
+ * 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.
@@ -16,16 +14,9 @@
* limitations under the License.
*/
-package net.openhft.chronicle.map;
+package net.openhft.chronicle.hash.locks;
-/**
- * @author Rob Austin.
- */
-public interface ReadContext extends Context {
-
- /**
- * @return true if the entry was present
- */
- boolean present();
+public interface LockingStrategy {
+ void lock(InterProcessLock lock);
}
diff --git a/src/main/java/net/openhft/chronicle/hash/replication/TimeProvider.java b/src/main/java/net/openhft/chronicle/hash/replication/TimeProvider.java
index 8aad67c91..3703a6147 100644
--- a/src/main/java/net/openhft/chronicle/hash/replication/TimeProvider.java
+++ b/src/main/java/net/openhft/chronicle/hash/replication/TimeProvider.java
@@ -27,7 +27,7 @@ import java.io.Serializable;
/**
* TimeProvider was aims to possibly later provide an optimization to
- * {@link System#currentTimeMillis()} on every call to replicated {@link ChronicleMap}
+ * {@link System#currentTime()} on every call to replicated {@link ChronicleMap}
* or {@link ChronicleSet}.
*
* Subclasses should be immutable, because {@link ChronicleHashBuilder} doesn't make defensive
@@ -38,17 +38,17 @@ import java.io.Serializable;
*/
public abstract class TimeProvider implements Serializable {
/**
- * Delegates {@link #currentTimeMillis()} to {@link System#currentTimeMillis()}.
+ * Delegates {@link #currentTime()} to {@link System#currentTime()}.
*/
public static final TimeProvider SYSTEM = new System();
private static final long serialVersionUID = 0L;
- public abstract long currentTimeMillis();
+ public abstract long currentTime();
private static class System extends TimeProvider {
private static final long serialVersionUID = 0L;
- public long currentTimeMillis() {
+ public long currentTime() {
return java.lang.System.currentTimeMillis();
}
diff --git a/src/main/java/net/openhft/chronicle/hash/serialization/internal/ByteableMarshaller.java b/src/main/java/net/openhft/chronicle/hash/serialization/internal/ByteableMarshaller.java
index db3682f36..6ccbaa1e8 100644
--- a/src/main/java/net/openhft/chronicle/hash/serialization/internal/ByteableMarshaller.java
+++ b/src/main/java/net/openhft/chronicle/hash/serialization/internal/ByteableMarshaller.java
@@ -119,7 +119,11 @@ public abstract class ByteableMarshaller
public void write(Bytes bytes, E e) {
Bytes eBytes = e.bytes();
if (eBytes != null) {
- bytes.write(eBytes, e.offset(), size);
+ if (eBytes != bytes || bytes.position() != e.offset()) {
+ bytes.write(eBytes, e.offset(), size);
+ } else {
+ bytes.skip(size);
+ }
} else {
throw new NullPointerException("You are trying to write a byteable object of " +
e.getClass() + ", " +
@@ -142,7 +146,7 @@ public abstract class ByteableMarshaller
try {
if (toReuse == null)
toReuse = getInstance();
- setBytesAndOffset(toReuse, bytes);
+ toReuse.bytes(bytes, bytes.position());
bytes.skip(size);
return toReuse;
} catch (Exception ex) {
diff --git a/src/main/java/net/openhft/chronicle/hash/serialization/internal/DataValueBytesMarshallers.java b/src/main/java/net/openhft/chronicle/hash/serialization/internal/DataValueBytesMarshallers.java
index c12280146..fcf065a91 100644
--- a/src/main/java/net/openhft/chronicle/hash/serialization/internal/DataValueBytesMarshallers.java
+++ b/src/main/java/net/openhft/chronicle/hash/serialization/internal/DataValueBytesMarshallers.java
@@ -492,8 +492,7 @@ public final class DataValueBytesMarshallers {
sb.append(" if (toReuse == null)\n");
sb.append(" toReuse = getInstance();\n");
sb.append(" if (toReuse instanceof Byteable) {\n");
- sb.append(" " +
- "ByteableMarshaller.setBytesAndOffset(((Byteable) toReuse), bytes);\n");
+ sb.append(" ((Byteable) toReuse).bytes(bytes, bytes.position());\n");
sb.append(" bytes.skip(size);\n");
sb.append(" return toReuse;\n");
sb.append(" }\n");
diff --git a/src/main/java/net/openhft/chronicle/hash/serialization/SizeMarshallers.java b/src/main/java/net/openhft/chronicle/hash/serialization/internal/SizeMarshallers.java
similarity index 95%
rename from src/main/java/net/openhft/chronicle/hash/serialization/SizeMarshallers.java
rename to src/main/java/net/openhft/chronicle/hash/serialization/internal/SizeMarshallers.java
index 758c603e7..ce00a356d 100644
--- a/src/main/java/net/openhft/chronicle/hash/serialization/SizeMarshallers.java
+++ b/src/main/java/net/openhft/chronicle/hash/serialization/internal/SizeMarshallers.java
@@ -16,8 +16,9 @@
* limitations under the License.
*/
-package net.openhft.chronicle.hash.serialization;
+package net.openhft.chronicle.hash.serialization.internal;
+import net.openhft.chronicle.hash.serialization.SizeMarshaller;
import net.openhft.lang.io.Bytes;
import static net.openhft.lang.io.IOTools.stopBitLength;
diff --git a/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java b/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java
index e02ea8388..1bc126e26 100644
--- a/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java
+++ b/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java
@@ -19,6 +19,7 @@
package net.openhft.chronicle.map;
import net.openhft.chronicle.hash.replication.ThrottlingConfig;
+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;
@@ -75,8 +76,7 @@ abstract class AbstractChannelReplicator implements Closeable {
private final Throttler throttler;
volatile boolean isClosed = false;
- ThreadLocalCopies copies;
- VanillaChronicleMap.SegmentState segmentState;
+ ReplicatedChronicleMap.BytesReplicatedContext context;
AbstractChannelReplicator(String name, ThrottlingConfig throttlingConfig)
throws IOException {
@@ -215,8 +215,7 @@ abstract class AbstractChannelReplicator implements Closeable {
@Override
public void run() {
try {
- copies = VanillaChronicleMap.SegmentState.getCopies(null);
- segmentState = VanillaChronicleMap.SegmentState.get(copies);
+ context = VanillaContext.get(BytesReplicatedContextFactory.INSTANCE);
processEvent();
} catch (Exception e) {
LOG.error("", e);
@@ -259,10 +258,10 @@ abstract class AbstractChannelReplicator implements Closeable {
public void closeResources() {
isClosed = true;
+ // context.close();
executorService.shutdown();
closeables.closeQuietly();
- if (segmentState != null)
- segmentState.close();
+
}
private void dumpThreadStackTrace(long start) {
@@ -438,6 +437,7 @@ abstract class AbstractChannelReplicator implements Closeable {
}
+ @Override
public Bytes resizeBuffer(int size) {
if (LOG.isDebugEnabled())
@@ -466,6 +466,7 @@ abstract class AbstractChannelReplicator implements Closeable {
return in;
}
+ @Override
public boolean shouldBeIgnored(final Bytes entry, final int chronicleId) {
return !externalizable.identifierCheck(entry, chronicleId);
}
diff --git a/src/main/java/net/openhft/chronicle/map/AbstractChronicleMap.java b/src/main/java/net/openhft/chronicle/map/AbstractChronicleMap.java
new file mode 100644
index 000000000..7755c75ed
--- /dev/null
+++ b/src/main/java/net/openhft/chronicle/map/AbstractChronicleMap.java
@@ -0,0 +1,388 @@
+/*
+ * 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.KeyContext;
+import org.jetbrains.annotations.NotNull;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.*;
+
+import static java.util.Collections.emptyList;
+
+abstract class AbstractChronicleMap extends AbstractMap
+ implements ChronicleMap, Serializable {
+ private static final long serialVersionUID = 0L;
+
+ transient Set> entrySet;
+
+ @Override
+ public abstract VanillaContext context(K key);
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public final boolean containsKey(Object key) {
+ try (MapKeyContext c = context((K) key)) {
+ return c.containsKey();
+ }
+ }
+
+ @Override
+ public final 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
+ // between, that will break ConcurrentMap.put() atomicity guarantee. So, we acquire
+ // update lock from the start:
+ c.updateLock().lock();
+ V prevValue = prevValueOnPut(c);
+ c.put(value);
+ return prevValue;
+ }
+ }
+
+ V prevValueOnPut(MapKeyContext context) {
+ return context.getUsing(null);
+ }
+
+ @Override
+ public final V putIfAbsent(K key, V value) {
+ try (MapKeyContext c = context(key)) {
+ // putIfAbsent() shouldn't actually put most of the time,
+ // so check if the key is present under read lock first:
+ if (c.readLock().tryLock()) {
+ // c.get() returns cached value, that might be unexpected by user,
+ // so use getUsing(null) which surely creates a new value instance:
+ V currentValue = c.getUsing(null);
+ if (currentValue != null)
+ return currentValue;
+ // Key is absent
+ upgradeReadToUpdateLockWithUnlockingIfNeeded(c);
+ }
+ // Entry with this key might be put into the map before we acquired
+ // update lock (exclusive) at any time, so even if we successfully upgraded
+ // to update lock, we should check if the value is still absent again
+ V currentValue = c.getUsing(null);
+ if (currentValue != null)
+ return currentValue;
+ // Key is absent
+ c.put(value);
+ return null;
+ }
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public final V get(Object key) {
+ return getUsing((K) key, null);
+ }
+
+ @Override
+ public final V getUsing(K key, V usingValue) {
+ try (MapKeyContext c = context(key)) {
+ return c.getUsing(usingValue);
+ }
+ }
+
+ @Override
+ public final 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:
+ if (c.readLock().tryLock()) {
+ V value = c.getUsing(usingValue);
+ if (value != null)
+ return value;
+ // Key is absent
+ upgradeReadToUpdateLockWithUnlockingIfNeeded(c);
+ } else {
+ c.updateLock().lock();
+ }
+ // Entry with this key might be put into the map before we acquired
+ // update lock (exclusive) at any time, so even if we successfully upgraded
+ // to update lock, we should check if the value is still absent again
+ V value = c.getUsing(usingValue);
+ if (value != null)
+ return value;
+ // Key is absent
+ putDefaultValue(c);
+ return c.getUsing(usingValue);
+ }
+ }
+
+ void upgradeReadToUpdateLockWithUnlockingIfNeeded(KeyContext c) {
+ if (!c.updateLock().tryLock()) {
+ c.readLock().unlock();
+ c.updateLock().lock();
+ }
+ }
+
+ abstract void putDefaultValue(VanillaContext context);
+
+ @Override
+ public R getMapped(K key, @NotNull Function super V, R> function) {
+ try (MapKeyContext c = context(key)) {
+ return c.containsKey() ? function.apply(c.get()) : null;
+ }
+ }
+
+ @Override
+ public V putMapped(K key, @NotNull UnaryOperator unaryOperator) {
+ try (MapKeyContext c = context(key)) {
+ // putMapped() should find a value, update & put most of the time,
+ // so don't try to check key presence under read lock first,
+ // as in putIfAbsent()/acquireUsing(), start with update lock:
+ c.updateLock().lock();
+ if (c.containsKey()) {
+ V newValue = unaryOperator.update(c.get());
+ c.put(newValue);
+ return newValue;
+ } else {
+ return null;
+ }
+ }
+ }
+
+ @Override
+ public synchronized void getAll(File toFile) throws IOException {
+ JsonSerializer.getAll(toFile, this, emptyList());
+ }
+
+ @Override
+ public synchronized void putAll(File fromFile) throws IOException {
+ JsonSerializer.putAll(fromFile, this, emptyList());
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public final 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
+ // between, that will break ConcurrentMap.remove() atomicity guarantee. So, we acquire
+ // update lock from the start:
+ c.updateLock().lock();
+ V prevValue = prevValueOnRemove(c);
+ c.remove();
+ return prevValue;
+ }
+ }
+
+ V prevValueOnRemove(MapKeyContext context) {
+ return context.getUsing(null);
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public final boolean remove(Object key, Object value) {
+ if (value == null)
+ return false; // CHM compatibility; General ChronicleMap policy is to throw NPE
+ try (MapKeyContext c = context((K) key)) {
+ // remove(key, value) should find the entry & remove most of the time,
+ // so don't try to check key presence and value equivalence under read lock first,
+ // as in putIfAbsent()/acquireUsing(), start with update lock:
+ c.updateLock().lock();
+ return c.containsKey() && c.valueEqualTo((V) value) && c.remove();
+ }
+ }
+
+ @Override
+ public final V replace(K key, V value) {
+ try (MapKeyContext c = context(key)) {
+ // replace(key, value) should find the key & put the value most of the time,
+ // so don't try to check key presence under read lock first,
+ // as in putIfAbsent()/acquireUsing(), start with update lock:
+ c.updateLock().lock();
+ if (c.containsKey()) {
+ // c.get() returns cached value, that might be unexpected by user,
+ // so use getUsing(null) which surely creates a new value instance:
+ V prevValue = c.getUsing(null);
+ c.put(value);
+ return prevValue;
+ } else {
+ return null;
+ }
+ }
+ }
+
+ @Override
+ public final boolean replace(K key, V oldValue, V newValue) {
+ try (MapKeyContext c = context(key)) {
+ // replace(key, old, new) should find the entry & put new value most of the time,
+ // so don't try to check key presence and value equivalence under read lock first,
+ // as in putIfAbsent()/acquireUsing(), start with update lock:
+ c.updateLock().lock();
+ return c.containsKey() && c.valueEqualTo(oldValue) && c.put(newValue);
+ }
+ }
+
+ abstract int actualSegments();
+
+ abstract VanillaContext rawContext();
+
+ abstract VanillaContext mapContext();
+
+ @NotNull
+ @Override
+ public final Set> entrySet() {
+ return (entrySet != null) ? entrySet : (entrySet = newEntrySet());
+ }
+
+ Set> newEntrySet() {
+ return new EntrySet();
+ }
+
+
+ class EntryIterator implements Iterator>, HashLookup.EntryConsumer {
+ /**
+ * Very inefficient (esp. if segments are large), but CORRECT implementation
+ * TODO map.forEachEntry(Consumer>)
+ */
+ private final Thread ownerThread = Thread.currentThread();
+ private int segmentIndex = actualSegments() - 1;
+ private final Queue> entryBuffer = new ArrayDeque<>();
+ private VanillaContext context;
+ private Entry returned;
+
+ 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);
+ } finally {
+ context = null;
+ }
+ }
+ }
+
+ @Override
+ public void accept(long key, long pos) {
+ context.pos = pos;
+ context.initKeyFromPos();
+ if (!context.containsKey()) // for replicated map
+ return;
+ entryBuffer.add(new WriteThroughEntry(context.immutableKey(), context.getUsing(null)));
+ }
+
+ @Override
+ public boolean hasNext() {
+ checkSingleThreaded();
+ fillEntryBuffer();
+ return !entryBuffer.isEmpty();
+ }
+
+ @Override
+ public Entry next() {
+ checkSingleThreaded();
+ fillEntryBuffer();
+ if ((returned = entryBuffer.poll()) == null)
+ throw new NoSuchElementException();
+ return returned;
+ }
+
+ @Override
+ public void remove() {
+ checkSingleThreaded();
+ if (returned == null)
+ throw new IllegalStateException();
+ AbstractChronicleMap.this.remove(returned.getKey(), returned.getValue());
+ returned = null;
+ }
+ }
+
+ 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);
+ }
+ }
+
+ 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();
+ }
+ }
+}
diff --git a/src/main/java/net/openhft/chronicle/map/BigSegmentHeader.java b/src/main/java/net/openhft/chronicle/map/BigSegmentHeader.java
new file mode 100644
index 000000000..48c4a5f8a
--- /dev/null
+++ b/src/main/java/net/openhft/chronicle/map/BigSegmentHeader.java
@@ -0,0 +1,451 @@
+/*
+ * 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.locks.IllegalInterProcessLockStateException;
+
+import java.util.concurrent.TimeUnit;
+
+import static java.nio.ByteOrder.LITTLE_ENDIAN;
+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;
+
+ static final long LOCK_OFFSET = 0L; // 64-bit
+ static final long COUNT_WORD_OFFSET = LOCK_OFFSET;
+ static final long WAIT_WORD_OFFSET = LOCK_OFFSET + 4L;
+
+ static final int COUNT_WORD_SHIFT = nativeOrder() == LITTLE_ENDIAN ? 0 : 32;
+ static final int WAIT_WORD_SHIFT = 32 - COUNT_WORD_SHIFT;
+
+ static final int READ_BITS = 30;
+ static final int MAX_READ = (1 << READ_BITS) - 1;
+ static final int READ_MASK = MAX_READ;
+ static final int READ_PARTY = 1;
+
+ static final int UPDATE_BIT = 1 << READ_BITS;
+ static final int UPDATE_PARTY = READ_PARTY | UPDATE_BIT;
+ static final int WRITE_BIT = UPDATE_BIT << 1;
+ static final int WRITE_LOCKED_COUNT_WORD = UPDATE_PARTY | WRITE_BIT;
+
+ static final int MAX_WAIT = Integer.MAX_VALUE;
+ static final int WAIT_PARTY = 1;
+
+ 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;
+
+ @Override
+ public long size(long address) {
+ return UNSAFE.getInt(address + SIZE_OFFSET) & UNSIGNED_INT_MASK;
+ }
+
+ @Override
+ public void size(long address, long size) {
+ UNSAFE.putInt(address + SIZE_OFFSET, (int) size);
+ }
+
+ @Override
+ public long nextPosToSearchFrom(long address) {
+ return UNSAFE.getInt(address + NEXT_POS_TO_SEARCH_FROM_OFFSET) & UNSIGNED_INT_MASK;
+ }
+
+ @Override
+ public void nextPosToSearchFrom(long address, long nextPosToSearchFrom) {
+ UNSAFE.putInt(address + NEXT_POS_TO_SEARCH_FROM_OFFSET, (int) nextPosToSearchFrom);
+ }
+
+ private static long getLockWord(long address) {
+ return UNSAFE.getLongVolatile(null, address + LOCK_OFFSET);
+ }
+
+ private static boolean casLockWord(long address, long expected, long x) {
+ return UNSAFE.compareAndSwapLong(null, address + LOCK_OFFSET, expected, x);
+ }
+
+ private static int countWord(long lockWord) {
+ return (int) (lockWord >> COUNT_WORD_SHIFT);
+ }
+
+ private static int waitWord(long lockWord) {
+ return (int) (lockWord >> WAIT_WORD_SHIFT);
+ }
+
+ private static long lockWord(int countWord, int waitWord) {
+ return ((((long) countWord) & UNSIGNED_INT_MASK) << COUNT_WORD_SHIFT) |
+ ((((long) waitWord) & UNSIGNED_INT_MASK) << WAIT_WORD_SHIFT);
+ }
+
+ private static int getCountWord(long address) {
+ return UNSAFE.getIntVolatile(null, address + COUNT_WORD_OFFSET);
+ }
+
+ private static boolean casCountWord(long address, int expected, int x) {
+ return UNSAFE.compareAndSwapInt(null, address + COUNT_WORD_OFFSET, expected, x);
+ }
+
+ private static void putCountWord(long address, int countWord) {
+ UNSAFE.putOrderedInt(null, address + COUNT_WORD_OFFSET, countWord);
+ }
+
+ private static boolean writeLocked(int countWord) {
+ return countWord == WRITE_LOCKED_COUNT_WORD;
+ }
+
+ private static void checkWriteLocked(int countWord) {
+ if (countWord != WRITE_LOCKED_COUNT_WORD)
+ throw new IllegalInterProcessLockStateException("Expected write lock");
+ }
+
+ private static boolean updateLocked(int countWord) {
+ return (countWord & UPDATE_BIT) != 0;
+ }
+
+ private static void checkUpdateLocked(int countWord) {
+ if (countWord < UPDATE_PARTY) // i. e. if update bit is not set, or write bit is set
+ throw new IllegalInterProcessLockStateException("Expected update lock");
+ }
+
+ private static int readCount(int countWord) {
+ return countWord & READ_MASK;
+ }
+
+ private static void checkReadLocked(int countWord) {
+ if (countWord <= 0) // i. e. if read count == 0 or write bit (actually sign bit) is set
+ throw new IllegalInterProcessLockStateException("Expected read lock");
+ }
+
+ private static void checkReadCountForIncrement(int countWord) {
+ if (readCount(countWord) == MAX_READ) {
+ throw new IllegalInterProcessLockStateException(
+ "Lock count reached the limit of " + MAX_READ);
+ }
+ }
+
+ private static int getWaitWord(long address) {
+ return UNSAFE.getIntVolatile(null, address + WAIT_WORD_OFFSET);
+ }
+
+ private static boolean casWaitWord(long address, int expected, int x) {
+ return UNSAFE.compareAndSwapInt(null, address + WAIT_WORD_OFFSET, expected, x);
+ }
+
+ private static void checkWaitWordForIncrement(int waitWord) {
+ if (waitWord == MAX_WAIT) {
+ throw new IllegalInterProcessLockStateException(
+ "Wait count reached the limit of " + MAX_WAIT);
+ }
+ }
+
+ private static void checkWaitWordForDecrement(int waitWord) {
+ if (waitWord == 0) {
+ throw new IllegalInterProcessLockStateException(
+ "Wait count underflowed");
+ }
+ }
+
+ private static void writeExclusiveLockHolder(long address) {
+ UNSAFE.putLong(address + EXCLUSIVE_LOCK_HOLDER_THREAD_ID_OFFSET,
+ Thread.currentThread().getId());
+ }
+
+ private static void clearExclusiveLockHolder(long address) {
+ UNSAFE.putLong(address + EXCLUSIVE_LOCK_HOLDER_THREAD_ID_OFFSET, 0L);
+ }
+
+ /**
+ * For debugging and monitoring
+ */
+ static Thread exclusiveLockHolder(long address) {
+ long holderId = UNSAFE.getLong(address + EXCLUSIVE_LOCK_HOLDER_THREAD_ID_OFFSET);
+ if (holderId == 0L)
+ return null;
+ Thread[] threads = new Thread[Thread.activeCount()];
+ Thread.enumerate(threads);
+ for (Thread thread : threads) {
+ if (thread.getId() == holderId)
+ return thread;
+ }
+ return null;
+ }
+
+ @Override
+ public void readLock(long address) {
+ if (!tryReadLock(address, 2, TimeUnit.SECONDS)) {
+ throw new RuntimeException("Dead lock");
+ }
+ }
+
+ @Override
+ public void readLockInterruptibly(long address) {
+ readLock(address);
+ }
+
+ @Override
+ public boolean tryReadLock(long address) {
+ long lockWord = getLockWord(address);
+ int countWord = countWord(lockWord);
+ if (!writeLocked(countWord) && waitWord(lockWord) == 0) {
+ checkReadCountForIncrement(countWord);
+ if (casCountWord(address, countWord, countWord + READ_PARTY))
+ return true;
+ }
+ return false;
+ }
+
+ @Override
+ public boolean tryReadLock(long address, long time, TimeUnit unit) {
+ return tryReadLock(address) || tryReadLock0(address, time, unit);
+ }
+
+ private boolean tryReadLock0(long address, long time, TimeUnit unit) {
+ long end = System.nanoTime() + unit.toNanos(time);
+ do {
+ if (tryReadLock(address))
+ return true;
+ } while (System.nanoTime() <= end);
+ return false;
+ }
+
+ @Override
+ public boolean tryUpgradeReadToUpdateLock(long address) {
+ int countWord = getCountWord(address);
+ checkReadLocked(countWord);
+ return !updateLocked(countWord) &&
+ casCountWord(address, countWord, countWord - READ_PARTY + UPDATE_PARTY);
+ }
+
+ @Override
+ public boolean tryUpgradeReadToWriteLock(long address) {
+ int countWord = getCountWord(address);
+ checkReadLocked(countWord);
+ return countWord == READ_PARTY &&
+ casCountWord(address, READ_PARTY, WRITE_LOCKED_COUNT_WORD);
+ }
+
+ @Override
+ public void updateLock(long address) {
+ if (!tryUpdateLock(address, 2, TimeUnit.SECONDS)) {
+ throw new RuntimeException("Dead lock");
+ }
+ }
+
+ @Override
+ public void updateLockInterruptibly(long address) {
+ updateLock(address);
+ }
+
+ @Override
+ public boolean tryUpdateLock(long address) {
+ long lockWord = getLockWord(address);
+ int countWord = countWord(lockWord);
+ if (!updateLocked(countWord) && waitWord(lockWord) == 0) {
+ checkReadCountForIncrement(countWord);
+ if (casCountWord(address, countWord, countWord + UPDATE_PARTY)) {
+ writeExclusiveLockHolder(address);
+ return true;
+ }
+ }
+ return false;
+ }
+
+ @Override
+ public boolean tryUpdateLock(long address, long time, TimeUnit unit) {
+ return tryUpdateLock(address) || tryUpdateLock0(address, time, unit);
+ }
+
+ private boolean tryUpdateLock0(long address, long time, TimeUnit unit) {
+ long end = System.nanoTime() + unit.toNanos(time);
+ do {
+ if (tryUpdateLock(address))
+ return true;
+ } while (System.nanoTime() <= end);
+ return false;
+ }
+
+ @Override
+ public void writeLock(long address) {
+ if (!tryWriteLock(address, 2, TimeUnit.SECONDS)) {
+ throw new RuntimeException("Dead lock");
+ }
+ }
+
+ @Override
+ public void writeLockInterruptibly(long address) {
+ writeLock(address);
+ }
+
+ @Override
+ public boolean tryWriteLock(long address) {
+ if (getCountWord(address) == 0 && casCountWord(address, 0, WRITE_LOCKED_COUNT_WORD)) {
+ writeExclusiveLockHolder(address);
+ return true;
+ } else {
+ return false;
+ }
+ }
+
+ @Override
+ public boolean tryWriteLock(long address, long time, TimeUnit unit) {
+ return tryWriteLock(address) || tryWriteLock0(address, time, unit);
+ }
+
+ private boolean tryWriteLock0(long address, long time, TimeUnit unit) {
+ long end = System.nanoTime() + unit.toNanos(time);
+ registerWait(address);
+ do {
+ long lockWord = getLockWord(address);
+ int countWord = countWord(lockWord);
+ if (countWord == 0) {
+ int waitWord = waitWord(lockWord);
+ checkWaitWordForDecrement(waitWord);
+ if (casLockWord(address, lockWord,
+ lockWord(WRITE_LOCKED_COUNT_WORD, waitWord - WAIT_PARTY))) {
+ writeExclusiveLockHolder(address);
+ return true;
+ }
+ }
+ } while (System.nanoTime() <= end);
+ deregisterWait(address);
+ return false;
+ }
+
+ private static void registerWait(long address) {
+ while (true) {
+ int waitWord = getWaitWord(address);
+ checkWaitWordForIncrement(waitWord);
+ if (casWaitWord(address, waitWord, waitWord + WAIT_PARTY))
+ return;
+ }
+ }
+
+ private static void deregisterWait(long address) {
+ while (true) {
+ int waitWord = getWaitWord(address);
+ checkWaitWordForDecrement(waitWord);
+ if (casWaitWord(address, waitWord, waitWord - WAIT_PARTY))
+ return;
+ }
+ }
+
+ @Override
+ public void upgradeUpdateToWriteLock(long address) {
+ if (!tryUpgradeUpdateToWriteLock(address, 2, TimeUnit.SECONDS)) {
+ throw new RuntimeException("Dead lock");
+ }
+ }
+
+ @Override
+ public void upgradeUpdateToWriteLockInterruptibly(long address) {
+ upgradeUpdateToWriteLock(address);
+ }
+
+ @Override
+ public boolean tryUpgradeUpdateToWriteLock(long address) {
+ int countWord = getCountWord(address);
+ return checkExclusiveUpdateLocked(countWord) &&
+ casCountWord(address, countWord, WRITE_LOCKED_COUNT_WORD);
+ }
+
+ private static boolean checkExclusiveUpdateLocked(int countWord) {
+ checkUpdateLocked(countWord);
+ return countWord == UPDATE_PARTY;
+ }
+
+ @Override
+ public boolean tryUpgradeUpdateToWriteLock(long address, long time, TimeUnit unit) {
+ return tryUpgradeUpdateToWriteLock(address) ||
+ tryUpgradeUpdateToWriteLock0(address, time, unit);
+ }
+
+ private boolean tryUpgradeUpdateToWriteLock0(long address, long time, TimeUnit unit) {
+ long end = System.nanoTime() + unit.toNanos(time);
+ registerWait(address);
+ do {
+ long lockWord = getLockWord(address);
+ int countWord = countWord(lockWord);
+ if (checkExclusiveUpdateLocked(countWord)) {
+ int waitWord = waitWord(lockWord);
+ checkWaitWordForDecrement(waitWord);
+ if (casLockWord(address, lockWord,
+ lockWord(WRITE_LOCKED_COUNT_WORD, waitWord - WAIT_PARTY))) {
+ return true;
+ }
+ }
+ } while (System.nanoTime() <= end);
+ deregisterWait(address);
+ return false;
+ }
+
+ @Override
+ public void readUnlock(long address) {
+ while (true) {
+ int countWord = getCountWord(address);
+ checkReadLocked(countWord);
+ if (casCountWord(address, countWord, countWord - READ_PARTY))
+ return;
+ }
+ }
+
+ @Override
+ public void updateUnlock(long address) {
+ while (true) {
+ int countWord = getCountWord(address);
+ checkUpdateLocked(countWord);
+ if (casCountWord(address, countWord, countWord - UPDATE_PARTY)) {
+ clearExclusiveLockHolder(address);
+ return;
+ }
+ }
+ }
+
+ @Override
+ public void downgradeUpdateToReadLock(long address) {
+ while (true) {
+ int countWord = getCountWord(address);
+ checkUpdateLocked(countWord);
+ if (casCountWord(address, countWord, countWord ^ UPDATE_BIT)) {
+ clearExclusiveLockHolder(address);
+ return;
+ }
+ }
+ }
+
+ @Override
+ public void writeUnlock(long address) {
+ checkWriteLocked(getCountWord(address));
+ clearExclusiveLockHolder(address);
+ putCountWord(address, 0);
+ }
+
+ @Override
+ public void downgradeWriteToUpdateLock(long address) {
+ checkWriteLocked(getCountWord(address));
+ putCountWord(address, UPDATE_PARTY);
+ }
+
+ @Override
+ public void downgradeWriteToReadLock(long address) {
+ checkWriteLocked(getCountWord(address));
+ clearExclusiveLockHolder(address);
+ putCountWord(address, READ_PARTY);
+ }
+}
diff --git a/src/main/java/net/openhft/chronicle/map/BytesChronicleMap.java b/src/main/java/net/openhft/chronicle/map/BytesChronicleMap.java
new file mode 100644
index 000000000..13f066f05
--- /dev/null
+++ b/src/main/java/net/openhft/chronicle/map/BytesChronicleMap.java
@@ -0,0 +1,119 @@
+/*
+ * Copyright 2014 Higher Frequency Trading http://www.higherfrequencytrading.com
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package net.openhft.chronicle.map;
+
+import net.openhft.lang.io.Bytes;
+import org.jetbrains.annotations.NotNull;
+
+import java.io.File;
+
+public class BytesChronicleMap extends AbstractChronicleMap {
+
+ private final VanillaChronicleMap, ?, ?, ?, ?, ?> delegate;
+ TcpReplicator.TcpSocketChannelEntryWriter output;
+
+ public BytesChronicleMap(VanillaChronicleMap, ?, ?, ?, ?, ?> delegate) {
+ this.delegate = delegate;
+ }
+
+ @Override
+ void putDefaultValue(VanillaContext context) {
+ delegate.putDefaultValue(context);
+ }
+
+ @Override
+ int actualSegments() {
+ return delegate.actualSegments();
+ }
+
+ @Override
+ VanillaContext rawContext() {
+ return delegate.rawBytesContext();
+ }
+
+ @Override
+ VanillaContext mapContext() {
+ VanillaContext context = delegate.bytesMapContext();
+ context.output = output;
+ return context;
+ }
+
+ @Override
+ public File file() {
+ return delegate.file();
+ }
+
+ @Override
+ public long longSize() {
+ return delegate.longSize();
+ }
+
+ @Override
+ public VanillaContext context(Bytes key) {
+ VanillaContext context = delegate.bytesMapContext();
+ context.output = output;
+ context.initKey(key);
+ return context;
+ }
+
+ @Override
+ public Class keyClass() {
+ return Bytes.class;
+ }
+
+ @Override
+ public void close() {
+ delegate.close();
+ }
+
+ @NotNull
+ @Override
+ public MapKeyContext acquireContext(@NotNull Bytes key, @NotNull Bytes usingValue) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Bytes newValueInstance() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Bytes newKeyInstance() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Class valueClass() {
+ return Bytes.class;
+ }
+
+ final void putAll(Bytes entries) {
+ long numberOfEntries = entries.readStopBit();
+ long entryPosition = entries.position();
+ while (numberOfEntries-- > 0) {
+ long keySize = delegate.keySizeMarshaller.readSize(entries);
+ entries.skip(keySize);
+ long valueSize = delegate.valueSizeMarshaller.readSize(entries);
+ long nextEntryPosition = entries.position() + valueSize;
+ entries.position(entryPosition);
+ put(entries, entries);
+ entries.clear(); // because used as key, altering position and limit
+ entryPosition = nextEntryPosition;
+ entries.position(entryPosition);
+ }
+ }
+}
diff --git a/src/main/java/net/openhft/chronicle/map/BytesMapEventListener.java b/src/main/java/net/openhft/chronicle/map/BytesMapEventListener.java
deleted file mode 100644
index 745221023..000000000
--- a/src/main/java/net/openhft/chronicle/map/BytesMapEventListener.java
+++ /dev/null
@@ -1,54 +0,0 @@
-/*
- * 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.serialization.SizeMarshaller;
-import net.openhft.lang.io.Bytes;
-
-import java.io.Serializable;
-
-/**
- *
- *
Methods must not update {@code entry} state ({@linkplain Bytes#position(long) position},
- * {@linkplain Bytes#limit() limit}).
- *
Methods must not update entry bytes, except in metadata area
- *
From {@code metaDataPos} offset metadata area starts in the given {@code entry}, listener
- * should be itself aware of it's length (see {@link ChronicleMapBuilder#metaDataBytes(int)}).
- *
From {@code keyPos} offset key area in the given {@code entry}, serialized key size
- * (using {@linkplain ChronicleMapBuilder#keySizeMarshaller(SizeMarshaller)}), directly followed
- * by the serialized key itself.
- *
From {@code valuePos} offset value area in the given {@code entry}, serialized value size
- * (using {@linkplain ChronicleMapBuilder#keySizeMarshaller(SizeMarshaller)}), then (optionally)
- * alignment (see {@link ChronicleMapBuilder#entryAndValueAlignment(Alignment)}
- * ), followed by the serialized value itself.
- *
- */
-public abstract class BytesMapEventListener implements Serializable {
- private static final long serialVersionUID = 0L;
-
- public void onGetFound(Bytes entry, long metaDataPos, long keyPos, long valuePos) {
- // do nothing
- }
-
- public void onPut(Bytes entry, long metaDataPos, long keyPos, long valuePos, boolean added) {
- // do nothing
- }
-
- public void onRemove(Bytes entry, long metaDataPos, long keyPos, long valuePos) {
- // do nothing
- }
-}
diff --git a/src/main/java/net/openhft/chronicle/map/ChannelProvider.java b/src/main/java/net/openhft/chronicle/map/ChannelProvider.java
index 2b877a3a8..8e08abe66 100644
--- a/src/main/java/net/openhft/chronicle/map/ChannelProvider.java
+++ b/src/main/java/net/openhft/chronicle/map/ChannelProvider.java
@@ -25,7 +25,6 @@ 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.threadlocal.ThreadLocalCopies;
import org.jetbrains.annotations.NotNull;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -136,8 +135,7 @@ final class ChannelProvider implements Closeable {
@Override
public void readExternalEntry(
- @NotNull ThreadLocalCopies copies,
- @NotNull VanillaChronicleMap.SegmentState segmentState, @NotNull Bytes source) {
+ @NotNull ReplicatedChronicleMap.BytesReplicatedContext context, @NotNull Bytes source) {
channelDataLock.readLock().lock();
try {
final int chronicleId = (int) source.readStopBit();
@@ -145,7 +143,7 @@ final class ChannelProvider implements Closeable {
// this channel is has not currently been created so it updates will be ignored
if (channelEntryExternalizables[chronicleId] != null)
channelEntryExternalizables[chronicleId]
- .readExternalEntry(copies, segmentState, source);
+ .readExternalEntry(context, source);
} else
LOG.info("skipped entry with chronicleId=" + chronicleId + ", ");
} finally {
@@ -509,8 +507,7 @@ final class ChannelProvider implements Closeable {
@Override
public void readExternalEntry(
- @NotNull ThreadLocalCopies copies,
- @NotNull VanillaChronicleMap.SegmentState segmentState, @NotNull Bytes source) {
+ @NotNull ReplicatedChronicleMap.BytesReplicatedContext context, @NotNull Bytes source) {
messageHandler.onMessage(source);
}
};
diff --git a/src/main/java/net/openhft/chronicle/map/ChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ChronicleMap.java
index 461ba033d..b1509fcff 100755
--- a/src/main/java/net/openhft/chronicle/map/ChronicleMap.java
+++ b/src/main/java/net/openhft/chronicle/map/ChronicleMap.java
@@ -28,6 +28,8 @@ import org.jetbrains.annotations.NotNull;
import java.io.Closeable;
import java.io.File;
import java.io.IOException;
+import java.io.Serializable;
+import java.lang.Object;
import java.util.Map;
import java.util.concurrent.ConcurrentMap;
@@ -51,14 +53,10 @@ import java.util.concurrent.ConcurrentMap;
* @param the map key type
* @param the map value type
*/
-public interface ChronicleMap extends ConcurrentMap, ChronicleHash {
- /**
- * Returns the number of entries in this map.
- *
- * @return number of entries in this map
- * @see Map#size()
- */
- long longSize();
+public interface ChronicleMap extends ConcurrentMap, ChronicleHash {
+
+ @Override
+ MapKeyContext context(K key);
/**
* Returns the value to which the specified key is mapped, or {@code null} if this map contains
@@ -102,50 +100,6 @@ public interface ChronicleMap extends ConcurrentMap, ChronicleHash {
*/
V getUsing(K key, V usingValue);
- /**
- * The method is similar {@link V getUsing(K key, V usingValue);} but in addition read locks
- * the map segment to operations to be performed atomically. ( see the example below )
- *
- * Returns the ReadContext which holds a map segment lock and provides method to get the value
- * ( to which the specified key is mapped) atomically ( see example below for an explanation ),
- * read to the provided {@code value} object, if possible, or returns {@code null}, if this map
- * contains no mapping for the key.
- *
- *
If the specified key is present in the map, the readContext.value() uses the provided
- * {@code usingValue} object via value marshaller's {@link BytesMarshaller#read(Bytes, Object)
- * read(Bytes, value)} or value reader's {@link BytesReader#read(Bytes, long, Object)
- * read(Bytes, size, value)} method, depending on what deserialization strategy is configured on
- * the builder, using which this map was constructed. If the value deserializer is able to reuse
- * the given {@code usingValue} object, calling this method instead of {@link #get(Object)}
- * could help to reduce garbage creation.
- *
{@code
- * try (ReadContext rc = map.getUsingLocked(key, bond)) {
- * if (rc.present ()) { // check whether the key was present
- * long issueDate = bond.getIssueDate();
- * String symbol = bond.getSymbol();
- * // add your logic here ( the lock will ensure this bond can not be changed by another thread
- * )
- * }
- * } // the read lock is released here
- * }
- * To ensure that you can read the 'issueDate' and 'symbol' can be read atomically, these values
- * must be read while the segment lock is in place.
- *
- *
- *
The provided {@code value} object is allowed to be {@code null}, in this case
- *
- * @param key the key whose associated value is to be returned
- * @param usingValue the object to read value data in, if possible
- * @return the read context containing the value to which the specified key is mapped
- *
- * no mapping for the key
- * @see #get(Object)
- * @see #getUsing(Object, Object)
- * @see ChronicleMapBuilder#valueMarshaller(BytesMarshaller)
- */
- @NotNull
- ReadContext getUsingLocked(@NotNull K key, @NotNull V usingValue);
-
/**
* Acquire a value for a key, creating if absent.
*
@@ -186,7 +140,7 @@ public interface ChronicleMap extends ConcurrentMap, ChronicleHash {
V acquireUsing(@NotNull K key, V usingValue);
@NotNull
- WriteContext acquireUsingLocked(@NotNull K key, @NotNull V usingValue);
+ MapKeyContext acquireContext(@NotNull K key, @NotNull V usingValue);
/**
* Apply a mapping to the value returned by a key and return a result. A read lock is assumed.
@@ -209,11 +163,11 @@ public interface ChronicleMap extends ConcurrentMap, ChronicleHash {
V putMapped(K key, @NotNull UnaryOperator unaryOperator);
/**
- * Exports all the entries to a {@link java.io.File} storing them in JSON format, an attempt is
+ * Exports all the entries to a {@link File} storing them in JSON format, an attempt is
* made where possible to use standard java serialisation and keep the data human readable, data
* serialized using the custom serialises are converted to a binary format which is not human
- * readable but this is only done if the Keys or Values are not {@link java.io.Serializable}.
- * This method can be used in conjunction with {@link ChronicleMap#putAll(java.io.File)} and is
+ * readable but this is only done if the Keys or Values are not {@link Serializable}.
+ * This method can be used in conjunction with {@link ChronicleMap#putAll(File)} and is
* especially useful if you wish to import/export entries from one chronicle map into another.
* This import and export of the entries can be performed even when the versions of ChronicleMap
* differ. This method is not performant and as such we recommend it is not used in performance
@@ -222,21 +176,21 @@ public interface ChronicleMap extends ConcurrentMap, ChronicleHash {
* @param toFile the file to store all the entries to, the entries will be stored in JSON
* format
* @throws IOException its not possible store the data to {@code toFile}
- * @see ChronicleMap#putAll(java.io.File)
+ * @see ChronicleMap#putAll(File)
*/
void getAll(File toFile) throws IOException;
/**
- * Imports all the entries from a {@link java.io.File}, the {@code fromFile} must be created
- * using or the same format as {@link ChronicleMap#get(java.lang.Object)}, this method behaves
- * similar to {@link java.util.Map#put(java.lang.Object, java.lang.Object)} where existing
+ * Imports all the entries from a {@link File}, the {@code fromFile} must be created
+ * using or the same format as {@link ChronicleMap#get(Object)}, this method behaves
+ * similar to {@link Map#put(Object, Object)} where existing
* entries are overwritten. A write lock is only held while each individual entry is inserted
- * into the map, not over all the entries in the {@link java.io.File}
+ * into the map, not over all the entries in the {@link File}
*
* @param fromFile the file containing entries ( in JSON format ) which will be deserialized and
- * {@link java.util.Map#put(java.lang.Object, java.lang.Object)} into the map
+ * {@link Map#put(Object, Object)} into the map
* @throws IOException its not possible read the {@code fromFile}
- * @see ChronicleMap#getAll(java.io.File)
+ * @see ChronicleMap#getAll(File)
*/
void putAll(File fromFile) throws IOException;
@@ -244,26 +198,24 @@ public interface ChronicleMap extends ConcurrentMap, ChronicleHash {
* Creates an empty value instance, which can be used with the
* following methods :
*
- * {@link ChronicleMap#getUsing(java.lang.Object, java.lang.Object) }
- * {@link ChronicleMap#getUsingLocked(java.lang.Object, java.lang.Object) }
- * {@link ChronicleMap#acquireUsing(java.lang.Object, java.lang.Object) }
- * {@link ChronicleMap#acquireUsingLocked(java.lang.Object, java.lang.Object) }
+ * {@link ChronicleMap#getUsing(Object, Object) }
+ * {@link ChronicleMap#acquireUsing(Object, Object) }
+ * {@link ChronicleMap#acquireContext(Object, Object) }
*
* for example like this :
*
*
{@code
* V value = map.newValueInstance();
- * try (ReadContext rc = map.getUsingLocked(key, value)) {
+ * try (ReadMapContext rc = map.getUsingLocked(key, value)) {
* // add your logic here
* } // the read lock is released here
* }
*
*
* @return a new empty instance based on the Value type
- * @see ChronicleMap#getUsing(java.lang.Object, java.lang.Object)
- * @see ChronicleMap#getUsingLocked(java.lang.Object, java.lang.Object)
- * @see ChronicleMap#acquireUsing(java.lang.Object, java.lang.Object)
- * @see ChronicleMap#acquireUsingLocked(java.lang.Object, java.lang.Object)
+ * @see ChronicleMap#getUsing(Object, Object)
+ * @see ChronicleMap#acquireUsing(Object, Object)
+ * @see ChronicleMap#acquireContext(Object, Object)
*/
V newValueInstance();
@@ -271,10 +223,9 @@ public interface ChronicleMap extends ConcurrentMap, ChronicleHash {
* Creates an empty value instance, which can be used with the
* following methods :
*
- * {@link ChronicleMap#getUsing(java.lang.Object, java.lang.Object) }
- * {@link ChronicleMap#getUsingLocked(java.lang.Object, java.lang.Object) }
- * {@link ChronicleMap#acquireUsing(java.lang.Object, java.lang.Object) }
- * {@link ChronicleMap#acquireUsingLocked(java.lang.Object, java.lang.Object) }
+ * {@link ChronicleMap#getUsing(Object, Object) }
+ * {@link ChronicleMap#acquireUsing(Object, Object) }
+ * {@link ChronicleMap#acquireContext(Object, Object) }
*
* for example like this :
*
@@ -282,25 +233,19 @@ public interface ChronicleMap extends ConcurrentMap, ChronicleHash {
* K key = map.newKeyInstance();
* key.setMyStringField("some key");
*
- * try (ReadContext rc = map.getUsingLocked(key, value)) {
+ * try (ReadMapContext rc = map.getUsingLocked(key, value)) {
* // add your logic here
* } // the read lock is released here
* }
*
*
* @return a new empty instance based on the Key type
- * @see ChronicleMap#getUsing(java.lang.Object, java.lang.Object)
- * @see ChronicleMap#getUsingLocked(java.lang.Object, java.lang.Object)
- * @see ChronicleMap#acquireUsing(java.lang.Object, java.lang.Object)
- * @see ChronicleMap#acquireUsingLocked(java.lang.Object, java.lang.Object)
+ * @see ChronicleMap#getUsing(Object, Object)
+ * @see ChronicleMap#acquireUsing(Object, Object)
+ * @see ChronicleMap#acquireContext(Object, Object)
*/
K newKeyInstance();
- /**
- * @return the class of {@code }
- */
- Class keyClass();
-
/**
* @return the class of {@code }
*/
diff --git a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java
index 6c2946ee7..d248f7004 100755
--- a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java
+++ b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java
@@ -19,8 +19,6 @@
package net.openhft.chronicle.map;
import net.openhft.chronicle.hash.ChronicleHashBuilder;
-import net.openhft.chronicle.hash.ChronicleHashErrorListener;
-import net.openhft.chronicle.hash.ChronicleHashErrorListeners;
import net.openhft.chronicle.hash.ChronicleHashInstanceBuilder;
import net.openhft.chronicle.hash.replication.*;
import net.openhft.chronicle.hash.serialization.*;
@@ -32,11 +30,13 @@ import net.openhft.lang.Maths;
import net.openhft.lang.io.Bytes;
import net.openhft.lang.io.BytesStore;
import net.openhft.lang.io.DirectStore;
+import net.openhft.lang.io.NativeBytes;
import net.openhft.lang.io.serialization.*;
import net.openhft.lang.io.serialization.impl.AllocateInstanceObjectFactory;
import net.openhft.lang.io.serialization.impl.NewInstanceObjectFactory;
import net.openhft.lang.io.serialization.impl.VanillaBytesMarshallerFactory;
import net.openhft.lang.model.Byteable;
+import net.openhft.lang.model.DataValueClasses;
import net.openhft.lang.threadlocal.Provider;
import net.openhft.lang.threadlocal.ThreadLocalCopies;
import org.jetbrains.annotations.NotNull;
@@ -44,7 +44,6 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.*;
-import java.lang.reflect.Constructor;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.net.InetSocketAddress;
@@ -99,7 +98,7 @@ import static net.openhft.lang.model.DataValueGenerator.firstPrimitiveFieldType;
* @see ChronicleMap
* @see ChronicleSetBuilder
*/
-public final class ChronicleMapBuilder implements Cloneable,
+public final class ChronicleMapBuilder implements
ChronicleHashBuilder, ChronicleMapBuilder>,
MapBuilder>, Serializable {
@@ -107,17 +106,18 @@ public final class ChronicleMapBuilder implements Cloneable,
private static final int DEFAULT_KEY_OR_VALUE_SIZE = 120;
private static final long DEFAULT_ENTRIES = 1 << 20;
private static final int MAX_SEGMENTS = (1 << 30);
- private static final int MAX_SEGMENTS_TO_CHAISE_COMPACT_MULTI_MAPS = (1 << 20);
private static final Logger LOG =
LoggerFactory.getLogger(ChronicleMapBuilder.class.getName());
- private static final StringBuilder EMTRY_STRING_BUILDER = new StringBuilder();
+ private static final StringBuilder EMPTY_STRING_BUILDER = new StringBuilder();
private static final double UNDEFINED_DOUBLE_CONFIG = Double.NaN;
private static final int XML_SERIALIZATION = 1;
private static final int BINARY_SERIALIZATION = 2;
+ static final long RUNTIME_PAGE_SIZE = (long) NativeBytes.UNSAFE.pageSize();
+
private static boolean isDefined(double config) {
return !Double.isNaN(config);
}
@@ -143,11 +143,10 @@ public final class ChronicleMapBuilder implements Cloneable,
private int actualChunkSize = 0;
private int maxChunksPerEntry = -1;
private Alignment alignment = null;
- private long entries = -1;
- private long lockTimeOut = 20000;
+ private long entries = -1L;
+ private long lockTimeOut = 20000L;
private TimeUnit lockTimeOutUnit = TimeUnit.MILLISECONDS;
private int metaDataBytes = 0;
- private ChronicleHashErrorListener errorListener = ChronicleHashErrorListeners.logging();
private boolean putReturnsNull = false;
private boolean removeReturnsNull = false;
@@ -155,11 +154,8 @@ public final class ChronicleMapBuilder implements Cloneable,
private TimeProvider timeProvider = TimeProvider.SYSTEM;
private BytesMarshallerFactory bytesMarshallerFactory;
private ObjectSerializer objectSerializer;
- private MapEventListener eventListener = null;
- private BytesMapEventListener bytesEventListener = null;
private V defaultValue = null;
private DefaultValueProvider defaultValueProvider = null;
- private PrepareValueBytes prepareValueBytes = null;
private SingleChronicleHashReplication singleHashReplication = null;
private InetSocketAddress[] pushToAddresses;
@@ -171,7 +167,7 @@ public final class ChronicleMapBuilder implements Cloneable,
if (CharSequence.class == valueClass)
defaultValue = (V) "";
if (StringBuilder.class == valueClass)
- defaultValue = (V) EMTRY_STRING_BUILDER;
+ defaultValue = (V) EMPTY_STRING_BUILDER;
}
/**
@@ -421,7 +417,7 @@ public final class ChronicleMapBuilder implements Cloneable,
constantlySizedKeys() && valueBuilder.constantSizeEncodingSizeMarshaller()) {
long constantSizeBeforeAlignment = round(size);
if (constantlySizedValues()) {
- // see specialEntrySpaceOffset()
+ // see segmentEntrySpaceInnerOffset()
long totalDataSize = constantSizeBeforeAlignment + constantValueSize();
worstAlignment = (int) (alignment.alignAddr(totalDataSize) - totalDataSize);
} else {
@@ -455,7 +451,7 @@ public final class ChronicleMapBuilder implements Cloneable,
* 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 specialEntrySpaceOffset(boolean replicated) {
+ int segmentEntrySpaceInnerOffset(boolean replicated) {
if (!constantlySizedEntries())
return 0;
return (int) (constantValueSize() % valueAlignment().alignment());
@@ -643,12 +639,15 @@ public final class ChronicleMapBuilder implements Cloneable,
if (actualChunksPerSegment > 0)
return entriesPerSegment;
double averageChunksPerEntry = averageChunksPerEntry(replicated);
- if (entriesPerSegment * averageChunksPerEntry > MultiMapFactory.MAX_CAPACITY)
+ if (entriesPerSegment * averageChunksPerEntry > HashLookup.MAX_SEGMENT_CHUNKS)
throw new IllegalStateException("Max chunks per segment is " +
- MultiMapFactory.MAX_CAPACITY + " configured entries() and " +
+ HashLookup.MAX_SEGMENT_CHUNKS + " configured entries() and " +
"actualSegments() so that there should be " + entriesPerSegment +
" entries per segment, while average chunks per entry is " +
averageChunksPerEntry);
+ if (entriesPerSegment > HashLookup.MAX_SEGMENT_ENTRIES)
+ throw new IllegalStateException("shouldn't be more than " +
+ HashLookup.MAX_SEGMENT_ENTRIES + " entries per segment");
return entriesPerSegment;
}
@@ -781,12 +780,10 @@ public final class ChronicleMapBuilder implements Cloneable,
segments = Math.max(minSegments, segments);
return (int) segments;
}
- long shortMMapSegments = trySegments(MultiMapFactory.I16_MAX_CAPACITY,
- MAX_SEGMENTS_TO_CHAISE_COMPACT_MULTI_MAPS, replicated);
- // TODO why not try I24 multiMap?
- if (shortMMapSegments > 0L)
- return (int) shortMMapSegments;
- long intMMapSegments = trySegments(MultiMapFactory.MAX_CAPACITY, MAX_SEGMENTS, replicated);
+ long smallSegments = trySegments(HashLookup.SMALL_SEGMENT_CHUNKS, MAX_SEGMENTS, replicated);
+ if (smallSegments > 0L)
+ return (int) smallSegments;
+ long intMMapSegments = trySegments(HashLookup.MAX_SEGMENT_CHUNKS, MAX_SEGMENTS, replicated);
if (intMMapSegments > 0L)
return (int) intMMapSegments;
throw new IllegalStateException("Max segments is " + MAX_SEGMENTS + ", configured so much" +
@@ -809,30 +806,6 @@ public final class ChronicleMapBuilder implements Cloneable,
return segments <= 16 * 1024 ? 64 : 32;
}
- MultiMapFactory multiMapFactory(boolean replicated) {
- return MultiMapFactory.forCapacity(actualChunksPerSegment(replicated));
- }
-
- public ChronicleMapBuilder lockTimeOut(long lockTimeOut, TimeUnit unit) {
- this.lockTimeOut = lockTimeOut;
- lockTimeOutUnit = unit;
- return this;
- }
-
- long lockTimeOut(TimeUnit unit) {
- return unit.convert(lockTimeOut, lockTimeOutUnit);
- }
-
- @Override
- public ChronicleMapBuilder errorListener(ChronicleHashErrorListener errorListener) {
- this.errorListener = errorListener;
- return this;
- }
-
- ChronicleHashErrorListener errorListener() {
- return errorListener;
- }
-
@Override
public ChronicleMapBuilder putReturnsNull(boolean putReturnsNull) {
this.putReturnsNull = putReturnsNull;
@@ -853,8 +826,8 @@ public final class ChronicleMapBuilder implements Cloneable,
return removeReturnsNull;
}
- @Override
- public ChronicleMapBuilder metaDataBytes(int metaDataBytes) {
+ // hidden for initial release.
+ ChronicleMapBuilder metaDataBytes(int metaDataBytes) {
if (metaDataBytes < 0 || metaDataBytes > 255)
throw new IllegalArgumentException("MetaDataBytes must be [0..255] was " + metaDataBytes);
this.metaDataBytes = metaDataBytes;
@@ -882,7 +855,6 @@ public final class ChronicleMapBuilder implements Cloneable,
", entries=" + entries() +
", lockTimeOut=" + lockTimeOut + " " + lockTimeOutUnit +
", metaDataBytes=" + metaDataBytes() +
- ", errorListener=" + errorListener() +
", putReturnsNull=" + putReturnsNull() +
", removeReturnsNull=" + removeReturnsNull() +
", timeProvider=" + timeProvider() +
@@ -890,10 +862,8 @@ public final class ChronicleMapBuilder implements Cloneable,
", objectSerializer=" + pretty(objectSerializer) +
", keyBuilder=" + keyBuilder +
", valueBuilder=" + valueBuilder +
- ", eventListener=" + eventListener +
", defaultValue=" + defaultValue +
", defaultValueProvider=" + pretty(defaultValueProvider) +
- ", prepareValueBytes=" + pretty(prepareValueBytes) +
'}';
}
@@ -965,7 +935,7 @@ public final class ChronicleMapBuilder implements Cloneable,
@Override
public ChronicleMapBuilder keyMarshallers(
- @NotNull BytesWriter keyWriter, @NotNull BytesReader keyReader) {
+ @NotNull BytesWriter super K> keyWriter, @NotNull BytesReader keyReader) {
keyBuilder.writer(keyWriter);
keyBuilder.reader(keyReader);
return this;
@@ -983,7 +953,7 @@ public final class ChronicleMapBuilder implements Cloneable,
*/
@Override
public ChronicleMapBuilder keyDeserializationFactory(
- @NotNull ObjectFactory keyDeserializationFactory) {
+ @NotNull ObjectFactory extends K> keyDeserializationFactory) {
keyBuilder.factory(keyDeserializationFactory);
return this;
}
@@ -1036,11 +1006,11 @@ public final class ChronicleMapBuilder implements Cloneable,
/**
* Configures the marshaller used to serialize actual value sizes to off-heap memory in maps,
- * created by this builder.
Default value size marshaller is so-called {@linkplain
- * SizeMarshallers#stopBit() stop bit encoding marshalling}, unless {@link
+ * created by this builder.
+ *
+ *
Default value size marshaller is so-called "stop bit encoding" marshalling, unless {@link
* #constantValueSizeBySample(Object)} or the builder statically knows the value size is
- * constant -- {@link SizeMarshallers#constant(long)} or equivalents are used by default in
- * these cases.
+ * constant -- special constant size marshalling is used by default in these cases.
*
* @param valueSizeMarshaller the new marshaller, used to serialize actual value sizes to
* off-heap memory
@@ -1078,116 +1048,53 @@ public final class ChronicleMapBuilder implements Cloneable,
return this;
}
- public ChronicleMapBuilder eventListener(MapEventListener eventListener) {
- this.eventListener = eventListener;
- return this;
- }
-
- MapEventListener eventListener() {
- return eventListener;
- }
-
- public ChronicleMapBuilder bytesEventListener(BytesMapEventListener eventListener) {
- this.bytesEventListener = eventListener;
- return this;
- }
-
- BytesMapEventListener bytesEventListener() {
- return bytesEventListener;
- }
-
/**
* Specifies the value to be put for each key queried in {@link ChronicleMap#acquireUsing
- * acquireUsing()} method, if the key is absent in the map, created by this builder.
This
- * configuration overrides any previous {@link #defaultValueProvider(DefaultValueProvider)} and
- * {@link #prepareDefaultValueBytes(PrepareValueBytes)} configurations to this builder.
+ * acquireUsing()} method, if the key is absent in the map, created by this builder.
+ *
+ *
This configuration overrides any previous {@link #defaultValueProvider(
+ * DefaultValueProvider)} configuration to this builder.
*
* @param defaultValue the default value to be put to the map for absent keys during {@code
* acquireUsing()} calls
* @return this builder object back
* @see #defaultValueProvider(DefaultValueProvider)
- * @see #prepareDefaultValueBytes(PrepareValueBytes)
*/
public ChronicleMapBuilder defaultValue(V defaultValue) {
if (defaultValue == null)
throw new IllegalArgumentException("default ChronicleMap value couldn't be null");
this.defaultValue = defaultValue;
this.defaultValueProvider = null;
- this.prepareValueBytes = null;
return this;
}
/**
* Specifies the function to obtain a value for the key during {@link ChronicleMap#acquireUsing
- * acquireUsing()} calls, if the key is absent in the map, created by this builder.
This
- * configuration overrides any previous {@link #defaultValue(Object)} and {@link
- * #prepareDefaultValueBytes(PrepareValueBytes)} configurations to this builder.
+ * acquireUsing()} calls, if the key is absent in the map, created by this builder.
+ *
+ *
This configuration overrides any previous {@link #defaultValue(Object)} configuration
+ * to this builder.
*
* @param defaultValueProvider the strategy to obtain a default value by the absent key
* @return this builder object back
* @see #defaultValue(Object)
- * @see #prepareDefaultValueBytes(PrepareValueBytes)
*/
public ChronicleMapBuilder defaultValueProvider(
@NotNull DefaultValueProvider defaultValueProvider) {
this.defaultValueProvider = defaultValueProvider;
this.defaultValue = null;
- this.prepareValueBytes = null;
return this;
}
- /**
- * Configures the procedure which is called on the bytes, which later the returned value is
- * pointing to or deserialized from, if the key is absent, on {@link ChronicleMap#acquireUsing
- * acquireUsing()} call on maps, created by this builder. See {@link PrepareValueBytes} for more
- * information.
This method of value initialization on {@code acquireUsing()} calls is
- * allowed only if value size is constant. Otherwise you should use either {@link
- * #defaultValue(Object)} or {@link #defaultValueProvider(DefaultValueProvider)} methods.
- *
This configuration overrides any previous {@link #defaultValue(Object)} and {@link
- * #defaultValueProvider(DefaultValueProvider)} configurations to this builder.
The
- * default preparation callback zeroes out the value bytes.
- *
- * @param prepareValueBytes what to do with the value bytes before assigning them into the
- * {@link Byteable} value or deserializing the value from, to return
- * from {@code acquireUsing()} call
- * @return this builder back
- * @throws IllegalStateException is value size is not constant
- * @see PrepareValueBytes
- * @see #defaultValue(Object)
- * @see #defaultValueProvider(DefaultValueProvider)
- */
- public ChronicleMapBuilder prepareDefaultValueBytes(
- @NotNull PrepareValueBytes prepareValueBytes) {
- this.prepareValueBytes = prepareValueBytes;
- this.defaultValue = null;
- this.defaultValueProvider = null;
- checkPrepareValueBytesOnlyIfConstantValueSize();
- return this;
- }
-
- private void checkPrepareValueBytesOnlyIfConstantValueSize() {
- if (prepareValueBytes != null && !constantlySizedValues())
- throw new IllegalStateException("Prepare value bytes could be used only if " +
- "value size is constant");
- }
-
- PrepareValueBytesAsWriter prepareValueBytesAsWriter() {
- PrepareValueBytes prepareValueBytes = this.prepareValueBytes;
- if ((prepareValueBytes == null && defaultValueProvider() != null) ||
- !constantlySizedValues())
- return null;
- long constantValueSize = constantValueSize();
- if (prepareValueBytes == null)
- prepareValueBytes = new ZeroOutValueBytes<>(constantValueSize);
- return new PrepareValueBytesAsWriter<>(prepareValueBytes, constantValueSize);
- }
-
/**
* Non-public because should be called only after {@link #preMapConstruction(boolean)}
*/
DefaultValueProvider defaultValueProvider() {
if (defaultValueProvider != null)
return defaultValueProvider;
+ V defaultValue = this.defaultValue;
+ if (defaultValue == null)
+ defaultValue = zeroValue();
if (defaultValue == null)
return null;
Object originalValueWriter = valueBuilder.interop();
@@ -1198,7 +1105,26 @@ public final class ChronicleMapBuilder implements Cloneable,
copies = metaWriterProvider.getCopies(copies);
MetaBytesWriter metaValueWriter = metaWriterProvider.get(copies,
valueBuilder.metaInterop(), valueWriter, defaultValue);
- return new ConstantValueProvider(defaultValue, metaValueWriter, valueWriter);
+ return new ConstantValueProvider<>(defaultValue, metaValueWriter, valueWriter);
+ }
+
+ private V zeroValue() {
+ if (Byteable.class.isAssignableFrom(valueBuilder.eClass)) {
+ try {
+ Object v = NativeBytes.UNSAFE.allocateInstance(valueBuilder.eClass);
+ Byteable defaultValue = (Byteable) v;
+ defaultValue.bytes(DirectStore.allocate(defaultValue.maxSize()).bytes(), 0);
+ return (V) v;
+ } catch (InstantiationException e) {
+ return null;
+ }
+ } else {
+ try {
+ return DataValueClasses.newDirectInstance(valueBuilder.eClass);
+ } catch (Exception e) {
+ return null;
+ }
+ }
}
@Override
@@ -1251,7 +1177,7 @@ public final class ChronicleMapBuilder implements Cloneable,
ChronicleMap createWithFile(File file, SingleChronicleHashReplication singleHashReplication,
ReplicationChannel channel) throws IOException {
- pushingToMapEventListener();
+ // pushingToMapEventListener();
for (int i = 0; i < 10; i++) {
if (file.exists() && file.length() > 0) {
try (FileInputStream fis = new FileInputStream(file);
@@ -1350,7 +1276,7 @@ public final class ChronicleMapBuilder implements Cloneable,
ChronicleMap createWithoutFile(
SingleChronicleHashReplication singleHashReplication, ReplicationChannel channel) {
try {
- pushingToMapEventListener();
+ // pushingToMapEventListener();
VanillaChronicleMap map = newMap(singleHashReplication, channel);
map.warnOnWindows();
BytesStore bytesStore = new DirectStore(JDKObjectSerializer.INSTANCE,
@@ -1362,33 +1288,33 @@ public final class ChronicleMapBuilder implements Cloneable,
throw new AssertionError(e);
}
}
-
- private void pushingToMapEventListener() {
- if (pushToAddresses == null || pushToAddresses.length == 0) {
- return;
- }
- try {
- Class> pmel = Class.forName(
- "com.higherfrequencytrading.chronicle.enterprise.map.PushingMapEventListener");
- Constructor> constructor = pmel.getConstructor(ChronicleMap[].class);
- // create a stateless client for each address
- ChronicleMap[] statelessClients = new ChronicleMap[pushToAddresses.length];
- ChronicleMapBuilder cmb = clone();
- cmb.pushTo((InetSocketAddress[]) null);
- for (int i = 0; i < pushToAddresses.length; i++) {
- statelessClients[i] = ChronicleMapStatelessClientBuilder.of(pushToAddresses[i])
- .create();
- }
- eventListener =
- (MapEventListener) constructor.newInstance((Object) statelessClients);
- } catch (ClassNotFoundException e) {
- LoggerFactory.getLogger(getClass().getName())
- .warn("Chronicle Enterprise not found in the class path");
- } catch (Exception e) {
- LoggerFactory.getLogger(getClass().getName())
- .error("PushingMapEventListener failed to load", e);
- }
- }
+//
+// private void pushingToMapEventListener() {
+// if (pushToAddresses == null || pushToAddresses.length == 0) {
+// return;
+// }
+// try {
+// Class> pmel = Class.forName(
+// "com.higherfrequencytrading.chronicle.enterprise.map.PushingMapEventListener");
+// Constructor> constructor = pmel.getConstructor(ChronicleMap[].class);
+// // create a stateless client for each address
+// ChronicleMap[] statelessClients = new ChronicleMap[pushToAddresses.length];
+// ChronicleMapBuilder cmb = clone();
+// cmb.pushTo((InetSocketAddress[]) null);
+// for (int i = 0; i < pushToAddresses.length; i++) {
+// statelessClients[i] = ChronicleMapStatelessClientBuilder.of(pushToAddresses[i])
+// .create();
+// }
+// eventListener =
+// (MapEventListener) constructor.newInstance((Object) statelessClients);
+// } catch (ClassNotFoundException e) {
+// LoggerFactory.getLogger(getClass().getName())
+// .warn("Chronicle Enterprise not found in the class path");
+// } catch (Exception e) {
+// LoggerFactory.getLogger(getClass().getName())
+// .error("PushingMapEventListener failed to load", e);
+// }
+// }
private VanillaChronicleMap newMap(
SingleChronicleHashReplication singleHashReplication, ReplicationChannel channel)
@@ -1440,7 +1366,6 @@ public final class ChronicleMapBuilder implements Cloneable,
throw new IllegalStateException("No info about value size");
}
checkAlignmentOnlyIfValuesPossiblyReferenceOffHeap();
- checkPrepareValueBytesOnlyIfConstantValueSize();
checkActualChunksPerSegmentIsConfiguredOnlyIfOtherLowLevelConfigsAreManual();
checkActualChunksPerSegmentGreaterOrEqualToEntries();
}
diff --git a/src/main/java/net/openhft/chronicle/map/ConstantValueProvider.java b/src/main/java/net/openhft/chronicle/map/ConstantValueProvider.java
index f55bcaacb..478bc085b 100644
--- a/src/main/java/net/openhft/chronicle/map/ConstantValueProvider.java
+++ b/src/main/java/net/openhft/chronicle/map/ConstantValueProvider.java
@@ -18,6 +18,7 @@
package net.openhft.chronicle.map;
+import net.openhft.chronicle.hash.KeyContext;
import net.openhft.chronicle.hash.serialization.BytesReader;
import net.openhft.chronicle.hash.serialization.internal.MetaBytesWriter;
import net.openhft.lang.io.ByteBufferBytes;
@@ -48,7 +49,7 @@ final class ConstantValueProvider implements DefaultValueProvider {
}
@Override
- public V get(K key) {
+ public V get(KeyContext keyContext) {
return value;
}
}
diff --git a/src/main/java/net/openhft/chronicle/map/Context.java b/src/main/java/net/openhft/chronicle/map/Context.java
deleted file mode 100644
index 7a8c53beb..000000000
--- a/src/main/java/net/openhft/chronicle/map/Context.java
+++ /dev/null
@@ -1,46 +0,0 @@
-/*
- * 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;
-
-/**
- * @author Rob Austin.
- */
-public interface Context extends AutoCloseable {
-
- /**
- * call this to unlock the maps segment
- */
- void close();
-
- /**
- * Returns the key corresponding to this entry.
- *
- * @return the key corresponding to this entry
- */
- K key();
-
- /**
- * Returns the value corresponding to this entry.
- *
- * @return the value corresponding to this entry
- */
- V value();
-
-}
-
diff --git a/src/main/java/net/openhft/chronicle/map/DefaultValueProvider.java b/src/main/java/net/openhft/chronicle/map/DefaultValueProvider.java
index 7a2d5ea52..a0010e7b7 100644
--- a/src/main/java/net/openhft/chronicle/map/DefaultValueProvider.java
+++ b/src/main/java/net/openhft/chronicle/map/DefaultValueProvider.java
@@ -18,6 +18,8 @@
package net.openhft.chronicle.map;
+import net.openhft.chronicle.hash.KeyContext;
+
import java.io.Serializable;
/**
@@ -32,8 +34,8 @@ public interface DefaultValueProvider extends Serializable {
* Returns a value to be put during {@link ChronicleMap#acquireUsing(Object, Object)} call
* for the specified key, if it is absent in the map.
*
- * @param key key absent in the map
+ * @param keyContext key absent in the map
* @return value to be put for the specified key in the map
*/
- V get(K key);
+ V get(KeyContext keyContext);
}
diff --git a/src/main/java/net/openhft/chronicle/map/HashLookup.java b/src/main/java/net/openhft/chronicle/map/HashLookup.java
new file mode 100644
index 000000000..3f086a3c1
--- /dev/null
+++ b/src/main/java/net/openhft/chronicle/map/HashLookup.java
@@ -0,0 +1,292 @@
+/*
+ * Copyright 2014 Higher Frequency Trading http://www.higherfrequencytrading.com
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package net.openhft.chronicle.map;
+
+import net.openhft.lang.Maths;
+
+import static net.openhft.lang.MemoryUnit.BITS;
+import static net.openhft.lang.MemoryUnit.BYTES;
+import static net.openhft.lang.io.NativeBytes.UNSAFE;
+
+class HashLookup {
+
+ // 9 bits for value, at most 9 + 3 = 12 bits for key, and a room for 2 bit slot state is left
+ // to fit 3 bytes/slot
+ static final int SMALL_SEGMENT_CHUNKS = 512;
+
+ // to fit 64 bits per slot.
+ static final int MAX_SEGMENT_CHUNKS = 1 << 30;
+ static final int MAX_SEGMENT_ENTRIES = 1 << 29;
+
+ static int valueBits(long actualChunksPerSegment) {
+ return 64 - Long.numberOfLeadingZeros(actualChunksPerSegment - 1L);
+ }
+
+ 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
+ // and key comparison, that is quite high cost.
+ // between 8.0 and 16.0
+ minKeyBits += 3;
+ int actualEntryBits = (int) BYTES.align((long) (minKeyBits + valueBits), BITS);
+ // devote the rest bits for key
+ return actualEntryBits - valueBits;
+ }
+
+ static int entrySize(int keyBits, int valueBits) {
+ return (int) BYTES.alignAndConvert((long) (keyBits + valueBits), BITS);
+ }
+
+ static long capacityFor(long entriesPerSegment) {
+ if (entriesPerSegment < 0L)
+ throw new IllegalArgumentException("entriesPerSegment should be positive");
+ long capacity = Maths.nextPower2(entriesPerSegment, 64L);
+ if (((double) entriesPerSegment) / (double) capacity > 2./3.) {
+ // hash lookup shouldn't be too dense
+ capacity <<= 1L;
+ }
+ return capacity;
+ }
+
+ private static final long UNSET_KEY = 0L;
+ private static final long UNSET_ENTRY = 0L;
+
+ private static long mask(int bits) {
+ return (1L << bits) - 1L;
+ }
+
+ private long address;
+
+ private long capacityMask;
+
+ private int entrySize;
+ private long capacityMask2;
+
+ private int keyBits;
+ private long keyMask;
+ private long valueMask;
+ private long entryMask;
+
+ // search state
+ private long searchKey = UNSET_KEY;
+ private long searchStartPos = -1L;
+ private long searchPos = -1L;
+
+ void reuse(long address, long capacity, int entrySize, int keyBits, int valueBits) {
+ this.address = address;
+
+ this.capacityMask = capacity - 1L;
+
+ this.entrySize = entrySize;
+ this.capacityMask2 = capacityMask * entrySize;
+
+ this.keyBits = keyBits;
+ this.keyMask = mask(keyBits);
+ this.valueMask = mask(valueBits);
+ this.entryMask = mask(keyBits + valueBits);
+ }
+
+ private long indexToPos(long index) {
+ return index * entrySize;
+ }
+
+ private long maskUnsetKey(long key) {
+ return (key &= keyMask) != UNSET_KEY ? key : keyMask;
+ }
+
+ private void checkValueForPut(long value) {
+ assert (value & ~valueMask) == 0L : "Value out of range, was " + value;
+ }
+
+ private boolean empty(long entry) {
+ return (entry & entryMask) == UNSET_ENTRY;
+ }
+
+ private long key(long entry) {
+ return entry & keyMask;
+ }
+
+ private long value(long entry) {
+ return (entry >>> keyBits) & valueMask;
+ }
+
+ private long entry(long key, long value) {
+ return key | (value << keyBits);
+ }
+
+ private long pos(long key) {
+ return indexToPos(key & capacityMask);
+ }
+
+ private long step(long pos) {
+ return (pos += entrySize) <= capacityMask2 ? pos : 0L;
+ }
+
+ private long stepBack(long pos) {
+ return (pos -= entrySize) >= 0 ? pos : capacityMask2;
+ }
+
+ private long readEntry(long pos) {
+ return UNSAFE.getLong(address + pos);
+ }
+
+ private void writeEntry(long pos, long prevEntry, long key, long value) {
+ long entry = (prevEntry & ~entryMask) | entry(key, value);
+ UNSAFE.putLong(address + pos, entry);
+ }
+
+ private void writeEntryVolatile(long pos, long prevEntry, long key, long value) {
+ long entry = (prevEntry & ~entryMask) | entry(key, value);
+ UNSAFE.putLongVolatile(null, address + pos, entry);
+ }
+
+ private void writeEntry(long pos, long prevEntry, long anotherEntry) {
+ long entry = (prevEntry & ~entryMask) | (anotherEntry & entryMask);
+ UNSAFE.putLong(address + pos, entry);
+ }
+
+ private void clearEntry(long pos, long prevEntry) {
+ long entry = (prevEntry & ~entryMask);
+ UNSAFE.putLong(address + pos, entry);
+ }
+
+ public void init0(long key) {
+ key = maskUnsetKey(key);
+ searchKey = key;
+ searchStartPos = pos(key);
+ }
+
+ public boolean isInit() {
+ return searchKey != UNSET_KEY;
+ }
+
+ public void close0() {
+ searchKey = UNSET_KEY;
+ }
+
+ public void initSearch0() {
+ searchPos = searchStartPos;
+ }
+
+ public boolean isSearchInit() {
+ return searchPos >= 0L;
+ }
+
+ public void closeSearch0() {
+ searchPos = -1L;
+ }
+
+ public long nextPos() {
+ long pos = searchPos;
+ while (true) {
+ long entry = readEntry(pos);
+ if (empty(entry)) {
+ searchPos = pos;
+ return -1L;
+ }
+ pos = step(pos);
+ if (pos == searchStartPos)
+ break;
+ if (key(entry) == searchKey) {
+ searchPos = pos;
+ return value(entry);
+ }
+ }
+ throw new IllegalStateException("MultiMap is full, that most likely means you " +
+ "misconfigured entrySize/chunkSize, and entries tend to take less chunks than " +
+ "expected");
+ }
+
+ public void found() {
+ searchPos = stepBack(searchPos);
+ }
+
+ public void remove() {
+ // for support of patterns like context.remove().put()
+ searchPos = remove0();
+ }
+
+ private long remove0() {
+ long posToRemove = searchPos;
+ long entryToRemove = readEntry(searchPos);
+ long posToShift = posToRemove;
+ while (true) {
+ posToShift = step(posToShift);
+ long entryToShift = readEntry(posToShift);
+ if (empty(entryToShift))
+ break;
+ long insertPos = pos(key(entryToShift));
+ // the following condition essentially means circular permutations
+ // of three (r = posToRemove, s = posToShift, i = insertPos)
+ // positions are accepted:
+ // [...i..r...s.] or
+ // [...r..s...i.] or
+ // [...s..i...r.]
+ boolean cond1 = insertPos <= posToRemove;
+ boolean cond2 = posToRemove <= posToShift;
+ if ((cond1 && cond2) ||
+ // chain wrapped around capacity
+ (posToShift < insertPos && (cond1 || cond2))) {
+ writeEntry(posToRemove, entryToRemove, entryToShift);
+ posToRemove = posToShift;
+ entryToRemove = entryToShift;
+ }
+ }
+ clearEntry(posToRemove, entryToRemove);
+ return posToRemove;
+ }
+
+ public void put(long value) {
+ checkValueForPut(value);
+ writeEntry(searchPos, readEntry(searchPos), searchKey, value);
+ }
+
+ public void putVolatile(long value) {
+ checkValueForPut(value);
+ writeEntryVolatile(searchPos, readEntry(searchPos), searchKey, value);
+ }
+
+ public void clear() {
+ UNSAFE.setMemory(address, capacityMask2 + entrySize, (byte) 0);
+ }
+
+ @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(',');
+ }
+ });
+ 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);
+ if (!empty(entry))
+ action.accept(key(entry), value(entry));
+ }
+ }
+}
diff --git a/src/main/java/net/openhft/chronicle/map/Int24Int24MultiMap.java b/src/main/java/net/openhft/chronicle/map/Int24Int24MultiMap.java
deleted file mode 100644
index 214912be2..000000000
--- a/src/main/java/net/openhft/chronicle/map/Int24Int24MultiMap.java
+++ /dev/null
@@ -1,301 +0,0 @@
-/*
- * Copyright 2014 Higher Frequency Trading
- *
- * http://www.higherfrequencytrading.com
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package net.openhft.chronicle.map;
-
-import net.openhft.lang.collection.ATSDirectBitSet;
-import net.openhft.lang.collection.DirectBitSet;
-import net.openhft.lang.io.Bytes;
-import net.openhft.lang.io.DirectStore;
-
-import static net.openhft.lang.Maths.isPowerOf2;
-
-/**
- * Supports a simple interface for int -> int[] off heap.
- */
-class Int24Int24MultiMap implements MultiMap {
-
- static final long MAX_CAPACITY = (1L << 24);
-
- private static final long ENTRY_SIZE = 6L;
- private static final int UNSET_KEY = 0;
- private static final long MASK = 0xFFFFFFL;
- private static final long HASH_INSTEAD_OF_UNSET_KEY = MASK;
- private static final long UNSET_ENTRY = 0L;
-
- final DirectBitSet positions;
- private final int capacity;
- private final long capacityMask;
- private final long capacityInBytes;
- private final Bytes bytes;
-
- public Int24Int24MultiMap(long minCapacity) {
- long capacity = MultiMapFactory.multiMapCapacity(minCapacity);
- this.capacity = (int) capacity;
- capacityMask = capacity - 1L;
- capacityInBytes = indexToPos(capacity);
- bytes = DirectStore.allocateLazy(indexToPos(capacity)).bytes();
- positions = MultiMapFactory.newPositions(capacity);
- clear();
- }
-
- public Int24Int24MultiMap(Bytes multiMapBytes, Bytes multiMapBitSetBytes) {
- long capacity = multiMapBytes.capacity() / ENTRY_SIZE;
- assert isPowerOf2(capacity);
- assert (capacity / 2L) <= MAX_CAPACITY;
- this.capacity = (int) capacity;
- capacityMask = capacity - 1L;
- capacityInBytes = indexToPos(capacity);
- this.bytes = multiMapBytes;
- positions = new ATSDirectBitSet(multiMapBitSetBytes);
- }
-
- /**
- * @param minCapacity as in {@link #Int24Int24MultiMap(long)} constructor
- * @return size of {@link Bytes} to provide to {@link #Int24Int24MultiMap(Bytes, Bytes)}
- * constructor as the first argument
- */
- public static long sizeInBytes(long minCapacity) {
- return indexToPos(MultiMapFactory.multiMapCapacity(minCapacity));
- }
-
- private static long indexToPos(long index) {
- return index * ENTRY_SIZE;
- }
-
- private static long maskUnsetKey(long key) {
- return (key &= MASK) != UNSET_KEY ? key : HASH_INSTEAD_OF_UNSET_KEY;
- }
-
- private void checkValueForPut(long value) {
- assert (value & ~MASK) == 0L : "Value out of range, was " + value;
- assert positions.isClear(value) : "Shouldn't put existing value";
- }
-
- private void checkValueForRemove(long value) {
- assert (value & ~MASK) == 0L : "Value out of range, was " + value;
- assert positions.isSet(value) : "Shouldn't remove absent value";
- }
-
- private static long key(long entry) {
- return entry >>> 24;
- }
-
- private static long value(long entry) {
- return entry & MASK;
- }
-
- private static long entry(long key, long value) {
- return (key << 24) | (value & MASK);
- }
-
- private long pos(long key) {
- return indexToPos(key & capacityMask);
- }
-
- private long step(long pos) {
- long t;
- return (t = (pos += ENTRY_SIZE) - capacityInBytes) < 0L ? pos : t;
- }
-
- private long stepBack(long pos) {
- return (pos -= ENTRY_SIZE) >= 0L ? pos : pos + capacityInBytes;
- }
-
- @Override
- public void putPosition(long value) {
- checkValueForPut(value);
- positions.set(value);
- }
-
- @Override
- public void put(long key, long value) {
- key = maskUnsetKey(key);
- checkValueForPut(value);
- for (long pos = pos(key); ; pos = step(pos)) {
- long entry = bytes.readInt48(pos);
- if (entry == UNSET_ENTRY) {
- bytes.writeInt48(pos, entry(key, value));
- positions.set(value);
- return;
- }
- }
- }
-
- @Override
- public void removePosition(long value) {
- checkValueForRemove(value);
- positions.clear(value);
- }
-
- @Override
- public void remove(long key, long value) {
- key = maskUnsetKey(key);
- checkValueForRemove(value);
- long posToRemove;
- for (long pos = pos(key); ; pos = step(pos)) {
- long entry = bytes.readInt48(pos);
- if (key(entry) == key && value(entry) == value) {
- posToRemove = pos;
- break;
- }
- }
- positions.clear(value);
- removePos(posToRemove);
- }
-
- @Override
- public void replace(long key, long oldValue, long newValue) {
- key = maskUnsetKey(key);
- checkValueForRemove(oldValue);
- checkValueForPut(newValue);
- for (long pos = pos(key); ; pos = step(pos)) {
- long entry = bytes.readInt48(pos);
- if (key(entry) == key && value(entry) == oldValue) {
- positions.clear(oldValue);
- positions.set(newValue);
- bytes.writeInt48(pos, entry(key, newValue));
- return;
- }
- }
- }
-
- private void removePos(long posToRemove) {
- long posToShift = posToRemove;
- while (true) {
- posToShift = step(posToShift);
- long entryToShift = bytes.readInt48(posToShift);
- if (entryToShift == UNSET_ENTRY)
- break;
- long insertPos = pos(key(entryToShift));
- // the following condition essentially means circular permutations
- // of three (r = posToRemove, s = posToShift, i = insertPos)
- // positions are accepted:
- // [...i..r...s.] or
- // [...r..s...i.] or
- // [...s..i...r.]
- boolean cond1 = insertPos <= posToRemove;
- boolean cond2 = posToRemove <= posToShift;
- if ((cond1 && cond2) ||
- // chain wrapped around capacity
- (posToShift < insertPos && (cond1 || cond2))) {
- bytes.writeInt48(posToRemove, entryToShift);
- posToRemove = posToShift;
- }
- }
- bytes.writeInt48(posToRemove, UNSET_ENTRY);
- }
-
- /////////////////////
- // Stateful methods
-
- @Override
- public void startSearch(long key, SearchState searchStateToReuse) {
- key = maskUnsetKey(key);
- searchStateToReuse.searchPos = pos(key);
- searchStateToReuse.searchHash = key;
- searchStateToReuse.putAfterFailedSearch = false;
- }
-
- @Override
- public long nextPos(SearchState searchState) {
- long pos = searchState.searchPos;
- while (true) {
- long entry = bytes.readInt48(pos);
- if (entry == UNSET_ENTRY) {
- searchState.searchPos = pos;
- return -1L;
- }
- pos = step(pos);
- if (key(entry) == searchState.searchHash) {
- searchState.searchPos = pos;
- return value(entry);
- }
- }
- }
-
- @Override
- public void removePrevPos(SearchState searchState) {
- long prevPos = stepBack(searchState.searchPos);
- long entry = bytes.readInt48(prevPos);
- positions.clear(value(entry));
- removePos(prevPos);
- }
-
- @Override
- public void replacePrevPos(SearchState searchState, long newValue,
- boolean oldValueInPositions) {
- checkValueForPut(newValue);
- long prevPos = searchState.searchPos;
- if (!searchState.putAfterFailedSearch)
- prevPos = stepBack(prevPos);
- if (oldValueInPositions) {
- long oldEntry = bytes.readInt48(prevPos);
- long oldValue = value(oldEntry);
- checkValueForRemove(oldValue);
- positions.clear(oldValue);
- }
- positions.set(newValue);
- bytes.writeInt48(prevPos, entry(searchState.searchHash, newValue));
- }
-
- @Override
- public void putAfterFailedSearch(SearchState searchState, long value) {
- checkValueForPut(value);
- positions.set(value);
- bytes.writeInt48(searchState.searchPos, entry(searchState.searchHash, value));
- }
-
- @Override
- public String toString() {
- StringBuilder sb = new StringBuilder();
- sb.append("{ ");
- long pos = 0L;
- for (int i = 0; i < capacity; i++, pos += ENTRY_SIZE) {
- long entry = bytes.readInt48(pos);
- if (entry != UNSET_ENTRY)
- sb.append(key(entry)).append('=').append(value(entry)).append(", ");
- }
- if (sb.length() > 2) {
- sb.setLength(sb.length() - 2);
- return sb.append(" }").toString();
- }
- return "{ }";
- }
-
- @Override
- public void forEach(EntryConsumer action) {
- long pos = 0L;
- for (int i = 0; i < capacity; i++, pos += ENTRY_SIZE) {
- long entry = bytes.readInt48(pos);
- if (entry != UNSET_ENTRY)
- action.accept(key(entry), value(entry));
- }
- }
-
- @Override
- public DirectBitSet getPositions() {
- return positions;
- }
-
- @Override
- public void clear() {
- positions.clear();
- bytes.zeroOut();
- }
-}
diff --git a/src/main/java/net/openhft/chronicle/map/IntIntMultiMap.java b/src/main/java/net/openhft/chronicle/map/IntIntMultiMap.java
deleted file mode 100644
index 5b0ee3b05..000000000
--- a/src/main/java/net/openhft/chronicle/map/IntIntMultiMap.java
+++ /dev/null
@@ -1,300 +0,0 @@
-/*
- * Copyright 2014 Higher Frequency Trading
- *
- * http://www.higherfrequencytrading.com
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package net.openhft.chronicle.map;
-
-import net.openhft.lang.collection.ATSDirectBitSet;
-import net.openhft.lang.collection.DirectBitSet;
-import net.openhft.lang.io.Bytes;
-import net.openhft.lang.io.DirectStore;
-
-import static net.openhft.chronicle.map.MultiMapFactory.multiMapCapacity;
-import static net.openhft.chronicle.map.MultiMapFactory.newPositions;
-import static net.openhft.lang.Maths.isPowerOf2;
-
-/**
- * Supports a simple interface for int -> int[] off heap.
- */
-class IntIntMultiMap implements MultiMap {
-
- static final long MAX_CAPACITY = (1L << 32);
-
- private static final long ENTRY_SIZE = 8L;
- private static final int ENTRY_SIZE_SHIFT = 3;
- private static final long UNSET_KEY = 0L;
- private static final long MASK = 0xFFFFFFFFL;
- private static final long HASH_INSTEAD_OF_UNSET_KEY = MASK;
- private static final long UNSET_ENTRY = 0L;
-
- final DirectBitSet positions;
- private final long capacity;
- private final long capacityMask;
- private final long capacityMask2;
- private final Bytes bytes;
-
- public IntIntMultiMap(long minCapacity) {
- capacity = multiMapCapacity(minCapacity);
- capacityMask = capacity - 1L;
- capacityMask2 = indexToPos(capacityMask);
- bytes = DirectStore.allocateLazy(indexToPos(capacity)).bytes();
- positions = newPositions(capacity);
- clear();
- }
-
- public IntIntMultiMap(Bytes multiMapBytes, Bytes multiMapBitSetBytes) {
- capacity = multiMapBytes.capacity() / ENTRY_SIZE;
- assert isPowerOf2(capacity);
- assert (capacity / 2L) <= MAX_CAPACITY;
- capacityMask = capacity - 1L;
- capacityMask2 = indexToPos(capacityMask);
- this.bytes = multiMapBytes;
- positions = new ATSDirectBitSet(multiMapBitSetBytes);
- }
-
- /**
- * @param minCapacity as in {@link #IntIntMultiMap(long)} constructor
- * @return size of {@link Bytes} to provide to {@link #IntIntMultiMap(Bytes, Bytes)}
- * constructor as the first argument
- */
- public static long sizeInBytes(long minCapacity) {
- return indexToPos(multiMapCapacity(minCapacity));
- }
-
- private static long indexToPos(long index) {
- return index << ENTRY_SIZE_SHIFT;
- }
-
- private static long maskUnsetKey(long key) {
- return (key &= MASK) != UNSET_KEY ? key : HASH_INSTEAD_OF_UNSET_KEY;
- }
-
- private void checkValueForPut(long value) {
- assert (value & ~MASK) == 0L : "Value out of range, was " + value;
- assert positions.isClear(value) : "Shouldn't put existing value";
- }
-
- private void checkValueForRemove(long value) {
- assert (value & ~MASK) == 0L : "Value out of range, was " + value;
- assert positions.isSet(value) : "Shouldn't remove absent value";
- }
-
- private static long key(long entry) {
- return entry >>> 32;
- }
-
- private static long value(long entry) {
- return entry & MASK;
- }
-
- private static long entry(long key, long value) {
- return (key << 32) | value;
- }
-
- private long pos(long key) {
- return indexToPos(key & capacityMask);
- }
-
- private long step(long pos) {
- return (pos + ENTRY_SIZE) & capacityMask2;
- }
-
- private long stepBack(long pos) {
- return (pos - ENTRY_SIZE) & capacityMask2;
- }
-
- @Override
- public void putPosition(long value) {
- checkValueForPut(value);
- positions.set(value);
- }
-
- @Override
- public void put(long key, long value) {
- key = maskUnsetKey(key);
- checkValueForPut(value);
- for (long pos = pos(key); ; pos = step(pos)) {
- long entry = bytes.readLong(pos);
- if (entry == UNSET_ENTRY) {
- bytes.writeLong(pos, entry(key, value));
- positions.set(value);
- return;
- }
- }
- }
-
- @Override
- public void removePosition(long value) {
- checkValueForRemove(value);
- positions.clear(value);
- }
-
- @Override
- public void remove(long key, long value) {
- key = maskUnsetKey(key);
- checkValueForRemove(value);
- long posToRemove;
- for (long pos = pos(key); ; pos = step(pos)) {
- long entry = bytes.readLong(pos);
- if (key(entry) == key && value(entry) == value) {
- posToRemove = pos;
- break;
- }
- }
- positions.clear(value);
- removePos(posToRemove);
- }
-
- @Override
- public void replace(long key, long oldValue, long newValue) {
- key = maskUnsetKey(key);
- checkValueForRemove(oldValue);
- checkValueForPut(newValue);
- for (long pos = pos(key); ; pos = step(pos)) {
- long entry = bytes.readLong(pos);
- if (key(entry) == key && value(entry) == oldValue) {
- positions.clear(oldValue);
- positions.set(newValue);
- bytes.writeLong(pos, entry(key, newValue));
- return;
- }
- }
- }
-
- private void removePos(long posToRemove) {
- long posToShift = posToRemove;
- while (true) {
- posToShift = step(posToShift);
- long entryToShift = bytes.readLong(posToShift);
- if (entryToShift == UNSET_ENTRY)
- break;
- long insertPos = pos(key(entryToShift));
- // the following condition essentially means circular permutations
- // of three (r = posToRemove, s = posToShift, i = insertPos)
- // positions are accepted:
- // [...i..r...s.] or
- // [...r..s...i.] or
- // [...s..i...r.]
- boolean cond1 = insertPos <= posToRemove;
- boolean cond2 = posToRemove <= posToShift;
- if ((cond1 && cond2) ||
- // chain wrapped around capacity
- (posToShift < insertPos && (cond1 || cond2))) {
- bytes.writeLong(posToRemove, entryToShift);
- posToRemove = posToShift;
- }
- }
- bytes.writeLong(posToRemove, UNSET_ENTRY);
- }
-
- /////////////////////
- // Stateful methods
-
- @Override
- public void startSearch(long key, SearchState searchStateToReuse) {
- key = maskUnsetKey(key);
- searchStateToReuse.searchPos = pos(key);
- searchStateToReuse.searchHash = key;
- searchStateToReuse.putAfterFailedSearch = false;
- }
-
- @Override
- public long nextPos(SearchState searchState) {
- long pos = searchState.searchPos;
- for(long count = capacity; count > 0; count--) {
- long entry = bytes.readLong(pos);
- if (entry == UNSET_ENTRY) {
- searchState.searchPos = pos;
- return -1L;
- }
- pos = step(pos);
- if (key(entry) == searchState.searchHash) {
- searchState.searchPos = pos;
- return value(entry);
- }
- }
- throw new IllegalStateException();
- }
-
- @Override
- public void removePrevPos(SearchState searchState) {
- long prevPos = stepBack(searchState.searchPos);
- long entry = bytes.readLong(prevPos);
- positions.clear(value(entry));
- removePos(prevPos);
- }
-
- @Override
- public void replacePrevPos(SearchState searchState, long newValue,
- boolean oldValueInPositions) {
- checkValueForPut(newValue);
- long prevPos = searchState.searchPos;
- if (!searchState.putAfterFailedSearch)
- prevPos = stepBack(prevPos);
- if (oldValueInPositions) {
- long oldEntry = bytes.readLong(prevPos);
- long oldValue = value(oldEntry);
- checkValueForRemove(oldValue);
- positions.clear(oldValue);
- }
- positions.set(newValue);
- bytes.writeLong(prevPos, entry(searchState.searchHash, newValue));
- }
-
- @Override
- public void putAfterFailedSearch(SearchState searchState, long value) {
- checkValueForPut(value);
- positions.set(value);
- bytes.writeLong(searchState.searchPos, entry(searchState.searchHash, value));
- }
-
- @Override
- public String toString() {
- StringBuilder sb = new StringBuilder();
- sb.append("{ ");
- for (long i = 0L, pos = 0L; i < capacity; i++, pos += ENTRY_SIZE) {
- long entry = bytes.readLong(pos);
- if (entry != UNSET_ENTRY)
- sb.append(key(entry)).append('=').append(value(entry)).append(", ");
- }
- if (sb.length() > 2) {
- sb.setLength(sb.length() - 2);
- return sb.append(" }").toString();
- }
- return "{ }";
- }
-
- @Override
- public void forEach(EntryConsumer action) {
- for (long i = 0L, pos = 0L; i < capacity; i++, pos += ENTRY_SIZE) {
- long entry = bytes.readLong(pos);
- if (entry != UNSET_ENTRY)
- action.accept(key(entry), value(entry));
- }
- }
-
- @Override
- public DirectBitSet getPositions() {
- return positions;
- }
-
- @Override
- public void clear() {
- positions.clear();
- bytes.zeroOut();
- }
-}
diff --git a/src/main/java/net/openhft/chronicle/map/MapEventListener.java b/src/main/java/net/openhft/chronicle/map/MapEventListener.java
deleted file mode 100644
index 8535d1b29..000000000
--- a/src/main/java/net/openhft/chronicle/map/MapEventListener.java
+++ /dev/null
@@ -1,135 +0,0 @@
-/*
- * 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.ChronicleHashBuilder;
-import org.jetbrains.annotations.Nullable;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.Serializable;
-
-/**
- * Contains methods which are called when {@link ChronicleMap} key events occur. Typical use cases:
- *
Map data backup / replication.
Logging, monitoring, debugging.
- *
- *
This is an adapter class - all methods have default implementations as no-ops. Extend this
- * class and override only methods corresponding the events you are interested in.
- *
- *
To configure {@code MapEventListener} for {@code ChronicleMap}, use {@link
- * ChronicleMapBuilder#eventListener(MapEventListener)} method.
- *
- *
See {@link #logging(String)} implementation.
- *
- *
All these calls are synchronous while a {@code ChronicleMap} lock is held so make them as
- * quick as possible.
- *
- *
The {@code entry} passed to the methods is {@code Bytes} instance positioned at meta data
- * area. See {@link ChronicleHashBuilder#metaDataBytes(int)} for more information.
- *
- * @param key type of the maps, trackable by this event listener
- * @param value type of the maps, trackable by this event listener
- * @see ChronicleMapBuilder#eventListener(MapEventListener)
- */
-public abstract class MapEventListener implements Serializable {
- private static final long serialVersionUID = 0L;
-
- /**
- * Returns the map event listener, which logs strings like "{@linkplain ChronicleMap#file() map
- * file} opType key value", where opType is either "get", "put" or "remove", to the logger
- * provided by SLF4J.
- *
- * @param the map key type
- * @param the map value type
- * @param prefix logging prefix
- * @return the logging event listener
- */
- public static MapEventListener logging(String prefix) {
- //noinspection unchecked
- return new LoggingMapEventListener(prefix);
- }
-
- private static class LoggingMapEventListener extends MapEventListener {
- private static final long serialVersionUID = 0L;
- public final static Logger LOGGER = LoggerFactory.getLogger(LoggingMapEventListener.class);
-
- private final String prefix;
-
- private LoggingMapEventListener(String prefix) {
- this.prefix = prefix;
- }
-
- @Override
- public void onGetFound(Object key, Object value) {
- LOGGER.info("get {} => {}", prefix, key, value);
- }
-
- @Override
- public void onPut(Object key, Object value, Object replacedValue) {
- LOGGER.info("{} put {} => {}", prefix, key, value);
- }
-
- @Override
- public void onRemove(Object key, Object value) {
- LOGGER.info("{} remove {} was {}", prefix, key, value);
- }
- }
-
- /**
- * This method is called if the key is found in the map during {@link ChronicleMap#get get},
- * {@link ChronicleMap#getUsing getUsing} or {@link ChronicleMap#acquireUsing acquireUsing}
- * method call.
- *
- * @param key the key looked up
- * @param foundValue the value found for the key
- */
- public void onGetFound(K key, V foundValue) {
- // do nothing
- }
-
- /**
- * This method is called whenever a new value is put for the key in the map during calls of such
- * methods as {@link ChronicleMap#put put}, {@link ChronicleMap#putIfAbsent putIfAbsent}, {@link
- * ChronicleMap#replace(Object, Object, Object) replace}, etc. When a new value is {@linkplain
- * ChronicleMapBuilder#defaultValue(Object) default} for the map or obtained during
- * {@link ChronicleMap#acquireUsing acquireUsing} call is put for the key, this method is called
- * as well.
- *
- *
This method is called when put is already happened.
- *
- * @param key the key the given value is put for
- * @param newValue the value which is now associated with the given key
- * @param replacedValue the value which was replaced by {@code newValue}, {@code null} if the
- * key was absent in the map before current {@code ChronicleMap}
- */
- public void onPut(K key, V newValue, @Nullable V replacedValue) {
- // do nothing
- }
-
- /**
- * This is called when an entry is removed. Misses, i. e. when {@code map.remove(key)} is
- * called, but key is already absent in the map, are not notified.
- *
- * @param key the key removed from the map
- * @param value the value which was associated with the given key
- */
- public void onRemove(K key, V value) {
- // do nothing
- }
-}
diff --git a/src/main/java/net/openhft/chronicle/map/GetValueInterops.java b/src/main/java/net/openhft/chronicle/map/MapKeyContext.java
similarity index 61%
rename from src/main/java/net/openhft/chronicle/map/GetValueInterops.java
rename to src/main/java/net/openhft/chronicle/map/MapKeyContext.java
index d0801b4d6..35f433135 100644
--- a/src/main/java/net/openhft/chronicle/map/GetValueInterops.java
+++ b/src/main/java/net/openhft/chronicle/map/MapKeyContext.java
@@ -16,12 +16,19 @@
package net.openhft.chronicle.map;
-import net.openhft.chronicle.hash.serialization.internal.MetaBytesInterop;
-import net.openhft.lang.threadlocal.ThreadLocalCopies;
-import org.jetbrains.annotations.NotNull;
+import net.openhft.chronicle.hash.KeyContext;
-interface GetValueInterops> {
- MVBI getMetaValueInterop(@NotNull ThreadLocalCopies copies, VBI valueInterop, VB value);
+public interface MapKeyContext extends KeyContext {
- VBI getValueInterop(@NotNull ThreadLocalCopies copies);
+ long valueOffset();
+
+ long valueSize();
+
+ boolean valueEqualTo(V value);
+
+ V get();
+
+ V getUsing(V usingValue);
+
+ boolean put(V newValue);
}
diff --git a/src/main/java/net/openhft/chronicle/map/MultiMap.java b/src/main/java/net/openhft/chronicle/map/MultiMap.java
deleted file mode 100644
index fd09274b1..000000000
--- a/src/main/java/net/openhft/chronicle/map/MultiMap.java
+++ /dev/null
@@ -1,79 +0,0 @@
-/*
- * Copyright 2014 Higher Frequency Trading http://www.higherfrequencytrading.com
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package net.openhft.chronicle.map;
-
-import net.openhft.lang.collection.DirectBitSet;
-
-/**
- * This is only used to store keys and positions, but it could store int/int key/values for another purpose.
- */
-interface MultiMap {
-
- /**
- * @return gets all the active positions as a bitset
- */
- DirectBitSet getPositions();
-
- /**
- * Add an entry. Allow duplicate hashes, but not key/position pairs.
- *
- * @param key to add
- * @param value to add
- */
- void put(long key, long value);
-
- void putPosition(long value);
-
- /**
- * Remove a key/value pair.
- *
- * @param key to remove
- * @param value to remove
- * @throws RuntimeException if key-value pair is absent in the multi map
- */
- void remove(long key, long value);
-
- void removePosition(long value);
-
- void replace(long key, long oldValue, long newValue);
-
- /**
- * Used for start a search for a given key
- */
- void startSearch(long key, SearchState searchStateToReuse);
-
- /**
- * Used for getting the next position for a given key
- *
- * @return the next position for the last search or negative value
- */
- long nextPos(SearchState searchState);
-
- void removePrevPos(SearchState searchState);
-
- void replacePrevPos(SearchState searchState, long newValue, boolean oldValueInPositions);
-
- void putAfterFailedSearch(SearchState searchState, long value);
-
- void clear();
-
- void forEach(EntryConsumer action);
-
- static interface EntryConsumer {
- void accept(long key, long value);
- }
-}
diff --git a/src/main/java/net/openhft/chronicle/map/MultiMapFactory.java b/src/main/java/net/openhft/chronicle/map/MultiMapFactory.java
deleted file mode 100644
index ca137ab23..000000000
--- a/src/main/java/net/openhft/chronicle/map/MultiMapFactory.java
+++ /dev/null
@@ -1,108 +0,0 @@
-/*
- * Copyright 2014 Higher Frequency Trading
- *
- * http://www.higherfrequencytrading.com
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package net.openhft.chronicle.map;
-
-import net.openhft.lang.Maths;
-import net.openhft.lang.collection.ATSDirectBitSet;
-import net.openhft.lang.io.DirectStore;
-import net.openhft.lang.io.NativeBytes;
-
-import static net.openhft.lang.Maths.isPowerOf2;
-import static net.openhft.lang.MemoryUnit.*;
-
-/**
- * Created by peter on 09/11/14.
- *
- * In future we might support I32 -> I16 or similar.
- */
-enum MultiMapFactory {
- I16 {
- @Override
- public long sizeInBytes(long minCapacity) {
- return ShortShortMultiMap.sizeInBytes(minCapacity);
- }
-
- @Override
- public MultiMap create(NativeBytes bytes, NativeBytes bitSetBytes) {
- return new ShortShortMultiMap(bytes, bitSetBytes);
- }
- },
- I24 {
- @Override
- public long sizeInBytes(long minCapacity) {
- return Int24Int24MultiMap.sizeInBytes(minCapacity);
- }
-
- @Override
- public MultiMap create(NativeBytes bytes, NativeBytes bitSetBytes) {
- return new Int24Int24MultiMap(bytes, bitSetBytes);
- }
- },
- I32 {
- @Override
- public long sizeInBytes(long minCapacity) {
- return IntIntMultiMap.sizeInBytes(minCapacity);
- }
-
- @Override
- public MultiMap create(NativeBytes bytes, NativeBytes bitSetBytes) {
- return new IntIntMultiMap(bytes, bitSetBytes);
- }
- };
-
- public static final long I16_MAX_CAPACITY = ShortShortMultiMap.MAX_CAPACITY;
- public static final long MAX_CAPACITY = IntIntMultiMap.MAX_CAPACITY;
-
- public static MultiMapFactory forCapacity(long capacity) {
- if (capacity <= ShortShortMultiMap.MAX_CAPACITY)
- return I16;
- // TODO why not I24?
-// if (capacity <= Int24Int24MultiMap.MAX_CAPACITY)
-// return I24;
- if (capacity <= IntIntMultiMap.MAX_CAPACITY)
- return I32;
- throw new IllegalArgumentException("Capacity " + capacity + " not supported");
- }
-
- public static long sizeOfBitSetInBytes(long actualChunksPerSegment) {
- return LONGS.align(BYTES.alignAndConvert(actualChunksPerSegment, BITS), BYTES);
- }
-
- public static ATSDirectBitSet newPositions(long capacity) {
- if (!isPowerOf2(capacity))
- throw new AssertionError("capacity should be a power of 2");
- capacity = LONGS.align(capacity, BITS);
- return new ATSDirectBitSet(DirectStore.allocateLazy(BYTES.convert(capacity, BITS)).bytes());
- }
-
- static long multiMapCapacity(long minCapacity) {
- if (minCapacity < 0L)
- throw new IllegalArgumentException("minCapacity should be positive");
- long capacity = Maths.nextPower2(minCapacity, 16L);
- if (((double) minCapacity) / capacity > 2./3.) {
- // multi map shouldn't be too dense
- capacity <<= 1L;
- }
- return capacity;
- }
-
- public abstract long sizeInBytes(long minCapacity);
-
- public abstract MultiMap create(NativeBytes bytes, NativeBytes bitSetBytes);
-}
diff --git a/src/main/java/net/openhft/chronicle/map/PrepareValueBytes.java b/src/main/java/net/openhft/chronicle/map/PrepareValueBytes.java
deleted file mode 100644
index a450891a8..000000000
--- a/src/main/java/net/openhft/chronicle/map/PrepareValueBytes.java
+++ /dev/null
@@ -1,61 +0,0 @@
-/*
- * Copyright 2014 Higher Frequency Trading http://www.higherfrequencytrading.com
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package net.openhft.chronicle.map;
-
-import net.openhft.lang.io.Bytes;
-import net.openhft.lang.model.Byteable;
-
-import java.io.Serializable;
-
-/**
- * Strategy of preparing bytes of off-heap memory before assigning them into {@link Byteable}
- * value during {@link ChronicleMap#acquireUsing(Object, Object)
- * chronicleMap.acquireUsing()} call, when the given key is absent in the map. Example: if you
- * want {@code 1L} to be the off-heap updatable ChronicleMap's {@linkplain
- * ChronicleMapBuilder#defaultValue(Object) default value}.
- * PrepareValueBytes) configure} custom {@code PrepareValueBytes} strategy in {@link
- * ChronicleMapBuilder}: