HCOLL-167 added support to delegate stateless client method calls to VChronicleMap

This commit is contained in:
Rob Austin
2014-10-28 21:29:34 +00:00
parent d4581ff985
commit f647f91b69
10 changed files with 263 additions and 189 deletions
@@ -900,7 +900,7 @@ public abstract class AbstractChronicleMapBuilder<K, V,
valueBuilder.constantSizeBySample(sampleValue);
}
private ChronicleMap<K, V> establishReplication(ChronicleMap<K, V> map)
private ChronicleMap<K, V> establishReplication(VanillaChronicleMap<K, ?, ?, V, ?, ?> map)
throws IOException {
if (map instanceof ReplicatedChronicleMap) {
ReplicatedChronicleMap result = (ReplicatedChronicleMap) map;
@@ -410,7 +410,8 @@ public final class ChannelProvider implements Closeable {
@Override
protected Closeable applyTo(AbstractChronicleMapBuilder builder,
Replica map, EntryExternalizable entryExternalizable, final ChronicleMap chronicleMap) {
Replica map, EntryExternalizable entryExternalizable,
final VanillaChronicleMap chronicleMap) {
if (builder.entrySize() > maxEntrySize) {
throw new IllegalArgumentException("During ReplicatingChannelBuilder setup, " +
"maxEntrySize=" + maxEntrySize + " was specified, but map with " +
@@ -49,6 +49,6 @@ abstract class Replicator {
*/
protected abstract Closeable applyTo(AbstractChronicleMapBuilder builder,
Replica map, Replica.EntryExternalizable entryExternalizable,
final ChronicleMap chronicleMap)
final VanillaChronicleMap chronicleMap)
throws IOException;
}
@@ -30,7 +30,7 @@ final class Replicators {
@Override
protected Closeable applyTo(AbstractChronicleMapBuilder builder,
Replica replica, Replica.EntryExternalizable entryExternalizable,
final ChronicleMap chronicleMap) throws IOException {
final VanillaChronicleMap chronicleMap) throws IOException {
final KeyValueSerializer keyValueSerializer = new KeyValueSerializer(builder
.keyBuilder, builder.valueBuilder);
@@ -50,7 +50,7 @@ final class Replicators {
@Override
protected Closeable applyTo(AbstractChronicleMapBuilder builder,
Replica map, Replica.EntryExternalizable entryExternalizable, final ChronicleMap chronicleMap)
Replica map, Replica.EntryExternalizable entryExternalizable, final VanillaChronicleMap chronicleMap)
throws IOException {
return new UdpReplicator(map, entryExternalizable, replicationConfig,
builder.entrySize());
@@ -85,7 +85,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
REMOVE_WITH_VALUE,
TO_STRING,
PUT_ALL,
EQUALS,
HASH_CODE
}
@@ -309,25 +309,23 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
* @return true if the contain the same data
*/
@Override
public boolean equals(Object object) {
public synchronized boolean equals(Object object) {
if (this == object) return true;
if (object == null || object.getClass().isAssignableFrom(Map.class))
return false;
final Map<? extends K, ? extends V> that = (Map<? extends K, ? extends V>) object;
long sizeLocation = writeEvent(EQUALS);
int size = size();
try {
writeEntries(that);
} catch (ClassCastException e) {
if (LOG.isDebugEnabled())
LOG.debug("failed equals() due to a difference in types", e);
if (that.size() != size)
return false;
}
// get the data back from the server
return blockingFetch(sizeLocation).readBoolean();
final Set<Map.Entry<K, V>> entries = entrySet();
return that.entrySet().equals(entries);
}
@@ -342,7 +340,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
@Override
public int hashCode() {
public synchronized int hashCode() {
long sizeLocation = writeEvent(HASH_CODE);
// get the data back from the server
@@ -1238,16 +1238,17 @@ class StatelessServerConnector<K, V> {
private static final Logger LOG = LoggerFactory.getLogger(StatelessServerConnector.class
.getName());
public static final int SIZE_OF_SIZE = 4;
public static final int TRANSACTION_ID_OFFSET = AbstractChannelReplicator.SIZE_OF_SIZE + 1;
public static final int TRANSACTIONID_OFFSET = AbstractChannelReplicator.SIZE_OF_SIZE + 1;
;
public static final boolean MAP_SUPPORTS_BYTES = Boolean.valueOf(System.getProperty
("mapSupportsBytes"));
private final KeyValueSerializer<K, V> keyValueSerializer;
private final ChronicleMap<K, V> map;
private final VanillaChronicleMap<K, ?, ?, V, ?, ?> map;
private double maxEntrySizeBytes;
StatelessServerConnector(@NotNull KeyValueSerializer<K, V> keyValueSerializer,
@NotNull ChronicleMap<K, V> map,
@NotNull VanillaChronicleMap<K, ?, ?, V, ?, ?> map,
int maxEntrySizeBytes) {
this.keyValueSerializer = keyValueSerializer;
this.map = map;
@@ -1304,7 +1305,6 @@ class StatelessServerConnector<K, V> {
case CLEAR:
return clear(reader, writer, sizeLocation);
case REPLACE:
return replace(reader, writer, sizeLocation);
@@ -1329,9 +1329,6 @@ class StatelessServerConnector<K, V> {
case HASH_CODE:
return hashCode(reader, writer, sizeLocation);
case EQUALS:
return equals(reader, writer, sizeLocation);
default:
throw new IllegalStateException("unsupported event=" + event);
@@ -1342,33 +1339,40 @@ class StatelessServerConnector<K, V> {
private Work removeWithValue(Bytes reader, Bytes writer, final long sizeLocation) {
if (MAP_SUPPORTS_BYTES) {
map.removeWithValue(reader, writer);
} else {
final K key = readKey(reader);
final V readValue = readValue(reader);
final K key = readKey(reader);
final V readValue = readValue(reader);
final boolean result;
try {
result = map.remove(key, readValue);
} catch (RuntimeException e) {
return sendException(writer, sizeLocation, e);
final boolean result;
try {
result = map.remove(key, readValue);
} catch (RuntimeException e) {
return sendException(writer, sizeLocation, e);
}
writer.writeBoolean(result);
}
writer.writeBoolean(result);
writeSizeAndFlags(sizeLocation, false, writer);
return null;
}
private Work replaceWithOldAndNew(Bytes reader, Bytes writer, final long sizeLocation) {
if (MAP_SUPPORTS_BYTES) {
map.replaceWithOldAndNew(reader, writer);
} else {
final K key = readKey(reader);
final V oldValue = readValue(reader);
final V newValue = readValue(reader);
boolean replaced;
final K key = readKey(reader);
final V oldValue = readValue(reader);
final V newValue = readValue(reader);
boolean replaced;
try {
replaced = map.replace(key, oldValue, newValue);
} catch (RuntimeException e) {
return sendException(writer, sizeLocation, e);
try {
replaced = map.replace(key, oldValue, newValue);
} catch (RuntimeException e) {
return sendException(writer, sizeLocation, e);
}
writer.writeBoolean(replaced);
}
writer.writeBoolean(replaced);
writeSizeAndFlags(sizeLocation, false, writer);
return null;
}
@@ -1397,13 +1401,14 @@ class StatelessServerConnector<K, V> {
}
private Work hashCode(ByteBufferBytes reader, Bytes writer, final long sizeLocation) {
private Work hashCode(Bytes reader, Bytes writer, final long sizeLocation) {
try {
writer.writeInt(map.hashCode());
} catch (RuntimeException e) {
return sendException(writer, sizeLocation, e);
}
writeSizeAndFlags(sizeLocation, false, writer);
return null;
}
@@ -1441,24 +1446,30 @@ class StatelessServerConnector<K, V> {
private Work isEmpty(Bytes reader, Bytes writer, final long sizeLocation) {
try {
writer.writeBoolean(map.isEmpty());
} catch (RuntimeException e) {
return sendException(writer, sizeLocation, e);
if (MAP_SUPPORTS_BYTES) {
map.isEmpty(reader, writer);
} else {
try {
writer.writeBoolean(map.isEmpty());
} catch (RuntimeException e) {
return sendException(writer, sizeLocation, e);
}
}
writeSizeAndFlags(sizeLocation, false, writer);
return null;
}
private Work containsKey(Bytes reader, Bytes writer, final long sizeLocation) {
if (MAP_SUPPORTS_BYTES) {
map.containsKey(reader, writer);
} else {
final K k = readKey(reader);
final K k = readKey(reader);
try {
writer.writeBoolean(map.containsKey(k));
} catch (RuntimeException e) {
return sendException(writer, sizeLocation, e);
try {
writer.writeBoolean(map.containsKey(k));
} catch (RuntimeException e) {
return sendException(writer, sizeLocation, e);
}
}
writeSizeAndFlags(sizeLocation, false, writer);
return null;
@@ -1466,27 +1477,33 @@ class StatelessServerConnector<K, V> {
private Work containsValue(Bytes reader, Bytes writer, final long sizeLocation) {
if (MAP_SUPPORTS_BYTES) {
map.containsValue(reader, writer);
} else {
final V v = readValue(reader);
final V v = readValue(reader);
try {
writer.writeBoolean(map.containsValue(v));
} catch (RuntimeException e) {
return sendException(writer, sizeLocation, e);
try {
writer.writeBoolean(map.containsValue(v));
} catch (RuntimeException e) {
return sendException(writer, sizeLocation, e);
}
}
writeSizeAndFlags(sizeLocation, false, writer);
return null;
}
private Work get(Bytes reader, Bytes writer, final long sizeLocation) {
if (MAP_SUPPORTS_BYTES) {
map.get(reader, writer);
} else {
final K k = readKey(reader);
final K k = readKey(reader);
try {
writeValue(map.get(k), writer);
} catch (RuntimeException e) {
return sendException(writer, sizeLocation, e);
try {
writeValue(map.get(k), writer);
} catch (RuntimeException e) {
return sendException(writer, sizeLocation, e);
}
}
writeSizeAndFlags(sizeLocation, false, writer);
return null;
@@ -1494,28 +1511,35 @@ class StatelessServerConnector<K, V> {
}
private Work put(Bytes reader, Bytes writer, final long sizeLocation) {
if (MAP_SUPPORTS_BYTES) {
map.put(reader, writer);
} else {
final K k = readKey(reader);
final V v = readValue(reader);
final K k = readKey(reader);
final V v = readValue(reader);
try {
writeValue(map.put(k, v), writer);
} catch (RuntimeException e) {
return sendException(writer, sizeLocation, e);
try {
writeValue(map.put(k, v), writer);
} catch (RuntimeException e) {
return sendException(writer, sizeLocation, e);
}
}
writeSizeAndFlags(sizeLocation, false, writer);
return null;
}
private Work remove(Bytes reader, Bytes writer, final long sizeLocation) {
final K key = readKey(reader);
if (MAP_SUPPORTS_BYTES) {
map.remove(reader, writer);
} else {
final K key = readKey(reader);
try {
writeValue(map.remove(key), writer);
} catch (RuntimeException e) {
return sendException(writer, sizeLocation, e);
}
try {
writeValue(map.remove(key), writer);
} catch (RuntimeException e) {
return sendException(writer, sizeLocation, e);
}
writeSizeAndFlags(sizeLocation, false, writer);
return null;
@@ -1524,43 +1548,31 @@ class StatelessServerConnector<K, V> {
private Work putAll(Bytes reader, Bytes writer, final long sizeLocation) {
if (MAP_SUPPORTS_BYTES) {
map.putAll(reader, writer);
} else {
final Map<K, V> m = readEntries(reader);
final Map<K, V> m = readEntries(reader);
try {
map.putAll(m);
} catch (RuntimeException e) {
return sendException(writer, sizeLocation, e);
}
try {
map.putAll(m);
} catch (RuntimeException e) {
return sendException(writer, sizeLocation, e);
}
writeSizeAndFlags(sizeLocation, false, writer);
return null;
}
private Work equals(Bytes reader, Bytes writer, final long sizeLocation) {
final Map<K, V> m = readEntries(reader);
boolean result;
try {
result = map.equals(m);
} catch (RuntimeException e) {
return sendException(writer, sizeLocation, e);
}
writer.writeBoolean(result);
writeSizeAndFlags(sizeLocation, false, writer);
return null;
}
private Work clear(Bytes reader, Bytes writer, final long sizeLocation) {
try {
map.clear();
} catch (RuntimeException e) {
return sendException(writer, sizeLocation, e);
}
writeSizeAndFlags(sizeLocation, false, writer);
return null;
}
@@ -1704,81 +1716,49 @@ class StatelessServerConnector<K, V> {
};
}
private Work sendException(Bytes reader, Bytes writer, RuntimeException e) {
final long sizeLocation = reflectTransactionId(reader, writer);
return sendException(writer, sizeLocation, e);
}
private long header(Bytes writer, final long transactionId) {
final long sizeLocation = writer.position();
writer.skip(TRANSACTION_ID_OFFSET); // SIZE_OF_SIZE + is
// exception
writer.writeLong(transactionId);
// hasAnotherChunk
writer.skip(1);
// count
writer.skip(4);
return sizeLocation;
}
private void writeHeader(Bytes writer, long sizeLocation, int count,
final boolean hasAnotherChunk) {
final long end = writer.position();
final int size = (int) (end - sizeLocation);
writer.position(sizeLocation);
// size in bytes
writer.writeInt(size);
// is exception
writer.writeBoolean(false);
//transaction id;
writer.skip(8);
writer.writeBoolean(hasAnotherChunk);
// count
writer.writeInt(count);
writer.position(end);
}
private Work putIfAbsent(Bytes reader, Bytes writer, final long sizeLocation) {
if (MAP_SUPPORTS_BYTES) {
map.putIfAbsent(reader, writer);
} else {
final K key = readKey(reader);
final V v = readValue(reader);
final K key = readKey(reader);
final V v = readValue(reader);
final V value;
final V result;
try {
value = map.putIfAbsent(key, v);
} catch (RuntimeException e) {
return sendException(writer, sizeLocation, e);
try {
result = map.putIfAbsent(key, v);
} catch (RuntimeException e) {
return sendException(writer, sizeLocation, e);
}
writeValue(result, writer);
}
writeValue(value, writer);
writeSizeAndFlags(sizeLocation, false, writer);
return null;
}
private Work replace(Bytes reader, Bytes writer, final long sizeLocation) {
final K k = readKey(reader);
final V v = readValue(reader);
final V replaced;
try {
replaced = map.replace(k, v);
} catch (RuntimeException e) {
return sendException(writer, sizeLocation, e);
if (MAP_SUPPORTS_BYTES) {
map.replaceKV(reader, writer);
} else {
final K k = readKey(reader);
final V v = readValue(reader);
final V replaced;
try {
replaced = map.replace(k, v);
} catch (RuntimeException e) {
return sendException(writer, sizeLocation, e);
}
writeValue(replaced, writer);
}
writeValue(replaced, writer);
writeSizeAndFlags(sizeLocation, false, writer);
return null;
}
@@ -1836,6 +1816,51 @@ class StatelessServerConnector<K, V> {
return result;
}
private Work sendException(Bytes reader, Bytes writer, RuntimeException e) {
final long sizeLocation = reflectTransactionId(reader, writer);
return sendException(writer, sizeLocation, e);
}
private long header(Bytes writer, final long transactionId) {
final long sizeLocation = writer.position();
writer.skip(TRANSACTION_ID_OFFSET); // SIZE_OF_SIZE + is
// exception
writer.writeLong(transactionId);
// hasAnotherChunk
writer.skip(1);
// count
writer.skip(4);
return sizeLocation;
}
private void writeHeader(Bytes writer, long sizeLocation, int count,
final boolean hasAnotherChunk) {
final long end = writer.position();
final int size = (int) (end - sizeLocation);
writer.position(sizeLocation);
// size in bytes
writer.writeInt(size);
// is exception
writer.writeBoolean(false);
//transaction id;
writer.skip(8);
writer.writeBoolean(hasAnotherChunk);
// count
writer.writeInt(count);
writer.position(end);
}
}
class KeyValueSerializer<K, V> {
@@ -54,7 +54,8 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
private static final Logger LOG = LoggerFactory.getLogger(VanillaChronicleMap.class);
/**
* Because DirectBitSet implementations couldn't find more than 64 continuous clear or set bits.
* Because DirectBitSet implementations couldn't find more than 64 continuous clear or set
* bits.
*/
static final int MAX_ENTRY_OVERSIZE_FACTOR = 64;
@@ -256,7 +257,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
long segmentSize() {
long ss = SEGMENT_HEADER
+ (CACHE_LINES.align(sizeOfMultiMap() + sizeOfMultiMapBitSet(), BYTES)
* multiMapsPerSegment())
* multiMapsPerSegment())
+ numberOfBitSets() * sizeOfBitSets() // the free list and 0+ dirty lists.
+ sizeOfEntriesInSegment();
if ((ss & 63L) != 0)
@@ -414,9 +415,9 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
/**
* removes ( if there exists ) an entry from the map, if the {@param key} and {@param expectedValue} match
* that of a maps.entry. If the {@param expectedValue} equals null then ( if there exists ) an entry whose
* key equals {@param key} this is removed.
* removes ( if there exists ) an entry from the map, if the {@param key} and {@param
* expectedValue} match that of a maps.entry. If the {@param expectedValue} equals null then (
* if there exists ) an entry whose key equals {@param key} this is removed.
*
* @param k the key of the entry to remove
* @param expectedValue null if not required
@@ -476,8 +477,8 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
* replace the value in a map, only if the existing entry equals {@param existingValue}
*
* @param key the key into the map
* @param existingValue the expected existing value in the map ( could be null when we don't wish to do
* this check )
* @param existingValue the expected existing value in the map ( could be null when we don't
* wish to do this check )
* @param newValue the new value you wish to store in the map
* @return the value that was replaced
*/
@@ -513,6 +514,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
}
static final ThreadLocal<MultiStoreBytes> tmpBytes = new ThreadLocal<>();
protected MultiStoreBytes acquireTmpBytes() {
MultiStoreBytes b = tmpBytes.get();
if (b == null)
@@ -520,6 +522,54 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
return b;
}
void replace(Bytes reader, Bytes writer) {
throw new UnsupportedOperationException("todo");
}
void isEmpty(Bytes reader, Bytes writer) {
throw new UnsupportedOperationException("todo");
}
void containsKey(Bytes reader, Bytes writer) {
throw new UnsupportedOperationException("todo");
}
void containsValue(Bytes reader, Bytes writer) {
throw new UnsupportedOperationException("todo");
}
void get(Bytes reader, Bytes writer) {
throw new UnsupportedOperationException("todo");
}
void putAll(Bytes reader, Bytes writer) {
throw new UnsupportedOperationException("todo");
}
void put(Bytes reader, Bytes writer) {
throw new UnsupportedOperationException("todo");
}
void putIfAbsent(Bytes reader, Bytes writer) {
throw new UnsupportedOperationException("todo");
}
void replaceKV(Bytes reader, Bytes writer) {
throw new UnsupportedOperationException("todo");
}
// for boolean replace(K key, V oldValue, V newValue);
void replaceWithOldAndNew(Bytes reader, Bytes writer) {
throw new UnsupportedOperationException("todo");
}
// for boolean remove(Object key, Object value);
void removeWithValue(Bytes reader, Bytes writer) {
throw new UnsupportedOperationException("todo");
}
// these methods should be package local, not public or private.
class Segment implements SharedSegment {
/*
@@ -551,7 +601,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
long start = bytes.startAddr() + SEGMENT_HEADER;
createHashLookups(start);
start += CACHE_LINES.align(sizeOfMultiMap() + sizeOfMultiMapBitSet(), BYTES)
* multiMapsPerSegment();
* multiMapsPerSegment();
final NativeBytes bsBytes = new NativeBytes(ms.objectSerializer(),
start,
start + LONGS.align(BYTES.alignAndConvert(entriesPerSegment, BITS), BYTES),
@@ -942,12 +992,12 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
/**
* Removes a key (or key-value pair) from the Segment.
*
* <p>The entry will only be removed if {@code expectedValue} equals to {@code null}
* or the value previously corresponding to the specified key.
* <p>The entry will only be removed if {@code expectedValue} equals to {@code null} or the
* value previously corresponding to the specified key.
*
* @param hash2 a hash code related to the {@code keyBytes}
* @return the value of the entry that was removed if the entry corresponding to the {@code keyBytes}
* exists and {@link #removeReturnsNull} is {@code false}, {@code null} otherwise
* @return the value of the entry that was removed if the entry corresponding to the {@code
* keyBytes} exists and {@link #removeReturnsNull} is {@code false}, {@code null} otherwise
*/
V remove(ThreadLocalCopies copies, MKI metaKeyInterop, KI keyInterop, K key,
V expectedValue, long hash2) {
@@ -1003,9 +1053,9 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
}
/**
* Replaces the specified value for the key with the given value. {@code newValue} is set only if
* the existing value corresponding to the specified key is equal to {@code expectedValue} or {@code
* expectedValue == null}.
* Replaces the specified value for the key with the given value. {@code newValue} is set
* only if the existing value corresponding to the specified key is equal to {@code
* expectedValue} or {@code expectedValue == null}.
*
* @param hash2 a hash code related to the {@code keyBytes}
* @return the replaced value or {@code null} if the value was not replaced
@@ -1081,17 +1131,17 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
}
/**
* Replaces value in existing entry. May cause entry relocation, because there may be not enough space
* for new value in location already allocated for this entry.
* Replaces value in existing entry. May cause entry relocation, because there may be not
* enough space for new value in location already allocated for this entry.
*
* @param pos index of the first block occupied by the entry
* @param offset relative offset of the entry in Segment bytes (before, i. e. including
* metaData)
* @param offset relative offset of the entry in Segment bytes (before, i. e.
* including metaData)
* @param entry relative pointer in Segment bytes
* @param valueSizePos relative position of value size in entry
* @param entryEndAddr absolute address of the entry end
* @return relative offset of the entry in Segment bytes after putting value (that may cause entry
* relocation)
* @return relative offset of the entry in Segment bytes after putting value (that may cause
* entry relocation)
*/
long putValue(long pos, long offset, NativeBytes entry, long valueSizePos,
long entryEndAddr, ThreadLocalCopies copies, V value, Bytes valueBytes,