Alawn/Chronicle-Map
1
0
Fork 0
mirror of https://github.com/OpenHFT/Chronicle-Map.git synced 2021-09-13 01:26:12 +00:00
Code Releases Activity

DOC-14 Stateless Client : enhancing the synchronous writes so we don't need a thread local client for each thread

Browse Source
This commit is contained in:
Rob Austin
2014-11-24 08:29:44 +00:00
parent 5fb243e301
commit a5fe252478
7 changed files with 92 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/map/ChronicleMap.java
+21 -11
View File
@@ -28,6 +28,7 @@ import org.jetbrains.annotations.NotNull;
import java.io.Closeable;
import java.util.Map;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.Future;
/**
* Extension of {@link ConcurrentMap} interface, stores the data off-heap.
@@ -165,11 +166,11 @@ public interface ChronicleMap<K, V> extends ConcurrentMap<K, V>, ChronicleHash {
* default value} or {@link AbstractChronicleMapBuilder#defaultValueProvider(DefaultValueProvider)
* defaultValueProvider.}
*
* <p>If the {@code ChronicleMap} is off-heap updatable, i. e. created via {@link
* ChronicleMapBuilder} builder (values are {@link Byteable}), there is one more option of
* what to do if the key is absent in the map, see {@link ChronicleMapBuilder#prepareValueBytesOnAcquire(PrepareValueBytes)}.
* By default, value bytes are just zeroed out, no default value, either provided for key or constant, is
* put for the absent key.
* <p>If the {@code ChronicleMap} is off-heap updatable, i. e. created via {@link ChronicleMapBuilder}
* builder (values are {@link Byteable}), there is one more option of what to do if the key is absent in
* the map, see {@link ChronicleMapBuilder#prepareValueBytesOnAcquire(PrepareValueBytes)}. By default,
* value bytes are just zeroed out, no default value, either provided for key or constant, is put for the
* absent key.
*
* @param key the key whose associated value is to be returned
* @param usingValue the object to read value data in, if present. Can not be null
@@ -183,8 +184,7 @@ public interface ChronicleMap<K, V> extends ConcurrentMap<K, V>, ChronicleHash {
WriteContext<K, V> acquireUsingLocked(@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.
* Apply a mapping to the value returned by a key and return a result. A read lock is assumed.
*
* @param key to apply the mapping to
* @param function to calculate a result
@@ -194,10 +194,8 @@ public interface ChronicleMap<K, V> extends ConcurrentMap<K, V>, ChronicleHash {
<R> R mapForKey(K key, @NotNull Function<? super V, R> function);
/**
* Apply a mutator to the value for a key and return a result.
* A write lock is assumed.
* <p>
* If there is no entry for this key it will be created and the value empty.
* Apply a mutator to the value for a key and return a result. A write lock is assumed. <p> If there is no
* entry for this key it will be created and the value empty.
*
* @param key to apply the mapping to
* @param mutator to alter the value and calculate a result
@@ -206,5 +204,17 @@ public interface ChronicleMap<K, V> extends ConcurrentMap<K, V>, ChronicleHash {
*/
<R> R updateForKey(K key, @NotNull Mutator<? super V, R> mutator);
/**
* Returns a {@link java.util.concurrent.Future} containing the value to which the specified key is
* mapped, or {@code null} if this map contains no mapping for the key. This method behaves the same as
* {@link java.util.Map#get(java.lang.Object)} yet wraps result in a future a {@link
* java.util.concurrent.Future}.
*
* @see java.util.Map#get(java.lang.Object)
* @see java.util.concurrent.Future
*/
Future<V> getLater(K key);
}
src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java
+6 -3
View File
@@ -37,6 +37,7 @@ import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.nio.channels.SocketChannel;
import java.util.*;
import java.util.concurrent.Future;
import static java.nio.ByteBuffer.allocateDirect;
import static net.openhft.chronicle.map.AbstractChannelReplicator.SIZE_OF_SIZE;
@@ -392,6 +393,10 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
return blockingFetch(writeEventAnSkip(LONG_SIZE)).readLong();
}
public synchronized Future<V> getLater(K key) {
throw new UnsupportedOperationException();
}
public synchronized V get(Object key) {
final long sizeLocation = writeEventAnSkip(GET);
@@ -951,8 +956,6 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
private Bytes blockingFetch(long timeoutTime, long transactionId) throws IOException {
// bytes.buffer().position(0);
// the number of bytes in the response
final int size = receive(SIZE_OF_SIZE, timeoutTime).readInt();
@@ -979,7 +982,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
", does not match the expected transaction-id=" + transactionId);
if (isException) {
Throwable throwable = (Throwable) bytes.readObject();
Throwable throwable = (Throwable) bytes.readObject();
try {
Field stackTrace = Throwable.class.getDeclaredField("stackTrace");
stackTrace.setAccessible(true);
src/main/java/net/openhft/chronicle/map/TcpReplicator.java
+14 -3
View File
@@ -40,6 +40,8 @@ import java.util.concurrent.atomic.AtomicBoolean;
import static java.nio.channels.SelectionKey.*;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static net.openhft.chronicle.map.AbstractChannelReplicator.SIZE_OF_SIZE;
import static net.openhft.chronicle.map.AbstractChannelReplicator.SIZE_OF_TRANSACTIONID;
import static net.openhft.chronicle.map.StatelessChronicleMap.EventId.HEARTBEAT;
interface Work {
@@ -1252,12 +1254,16 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
*/
class StatelessServerConnector<K, V> {
public static final int TRANSACTION_ID_OFFSET = AbstractChannelReplicator.SIZE_OF_SIZE + 1;
public static final int TRANSACTION_ID_OFFSET = SIZE_OF_SIZE + 1;
public static final boolean MAP_SUPPORTS_BYTES = Boolean.valueOf(System.getProperty
("mapSupportsBytes"));
private static final Logger LOG = LoggerFactory.getLogger(StatelessServerConnector.class
.getName());
public static final StatelessChronicleMap.EventId[] VALUES = StatelessChronicleMap.EventId.values();
public static final int SIZE_OF_IS_EXCEPTION = 1;
public static final int HEADER_SIZE = SIZE_OF_SIZE + SIZE_OF_IS_EXCEPTION + SIZE_OF_TRANSACTIONID;
private final KeyValueSerializer<K, V> keyValueSerializer;
private final VanillaChronicleMap<K, ?, ?, V, ?, ?> map;
private double maxEntrySizeBytes;
@@ -1500,7 +1506,12 @@ class StatelessServerConnector<K, V> {
private Work sendException(Bytes writer, long sizeLocation, Throwable e) {
// move the position to ignore any bytes written so far
writer.position(sizeLocation + HEADER_SIZE);
writeException(writer, e);
writeSizeAndFlags(sizeLocation, true, writer);
return null;
}
@@ -1898,8 +1909,8 @@ class StatelessServerConnector<K, V> {
private void writeSizeAndFlags(long locationOfSize, boolean isException, Bytes out) {
long currentPosition = out.position();
final long size = out.position() - locationOfSize;
out.writeInt(locationOfSize, (int) size);
out.writeBoolean(locationOfSize + AbstractChannelReplicator.SIZE_OF_SIZE, isException);
out.writeInt(locationOfSize, (int) size); // size
out.writeBoolean(locationOfSize + SIZE_OF_SIZE, isException); // isException
out.position(currentPosition);
}
src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java
+37 -2
View File
@@ -42,7 +42,10 @@ import java.io.Serializable;
import java.lang.reflect.Array;
import java.nio.channels.FileChannel;
import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import static java.lang.Long.numberOfTrailingZeros;
import static java.lang.Math.max;
@@ -80,7 +83,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
final Alignment alignment;
final int actualSegments;
final long entriesPerSegment;
final MapEventListener<K, V, ChronicleMap<K, V>> eventListener;
final transient MapEventListener<K, V, ChronicleMap<K, V>> eventListener;
// if set the ReturnsNull fields will cause some functions to return NULL
// rather than as returning the Object can be expensive for something you probably don't use.
final boolean putReturnsNull;
@@ -314,7 +317,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
void checkValue(Object value) {
if (vClass != Void.class && !vClass.isInstance(value)) {
throw new ClassCastException("Value must be a " + vClass.getName() +
" but was a "+ value.getClass());
" but was a " + value.getClass());
}
}
@@ -381,6 +384,38 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
}
}
@Override
public Future<V> getLater(K key) {
final V v = get(key);
return new Future<V>() {
@Override
public boolean cancel(boolean mayInterruptIfRunning) {
return false;
}
@Override
public boolean isCancelled() {
return false;
}
@Override
public boolean isDone() {
return true;
}
@Override
public V get() throws InterruptedException, ExecutionException {
return v;
}
@Override
public V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
return v;
}
};
}
@NotNull
@Override
public WriteContext<K, V> acquireUsingLocked(@NotNull K key, @NotNull V usingValue) {