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

Enhanced thread safety tests with the latest lang release

Browse Source
This commit is contained in:
Peter Lawrey
2014-10-27 15:59:22 +00:00
parent 33d37c493f
commit a956ac2a0e
12 changed files with 102 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/common/serialization/impl/ByteableMarshaller.java
+13 -1
View File
@@ -21,6 +21,7 @@ package net.openhft.chronicle.common.serialization.impl;
import net.openhft.chronicle.common.serialization.AgileBytesMarshaller;
import net.openhft.chronicle.common.serialization.Hasher;
import net.openhft.lang.io.Bytes;
import net.openhft.lang.io.MultiStoreBytes;
import net.openhft.lang.io.NativeBytes;
import net.openhft.lang.io.serialization.*;
import net.openhft.lang.io.serialization.impl.AllocateInstanceObjectFactory;
@@ -117,7 +118,7 @@ public class ByteableMarshaller<E extends Byteable> implements AgileBytesMarshal
try {
if (e == null)
e = getInstance();
e.bytes(bytes, bytes.position());
setBytesAndOffset(e, bytes);
bytes.skip(size);
return e;
} catch (Exception ex) {
@@ -125,6 +126,17 @@ public class ByteableMarshaller<E extends Byteable> implements AgileBytesMarshal
}
}
public static void setBytesAndOffset(Byteable e, Bytes bytes) {
// use the unshifting underlying Bytes (the position never changes)
// if you use a Bytes where the position changes, you risk concurrency issues.
if (bytes instanceof MultiStoreBytes) {
MultiStoreBytes msb = (MultiStoreBytes) bytes;
e.bytes(msb.underlyingBytes(), msb.underlyingOffset() + msb.position());
} else {
e.bytes(bytes, bytes.position());
}
}
@SuppressWarnings("unchecked")
@NotNull
E getInstance() throws Exception {
src/main/java/net/openhft/chronicle/map/NodeDiscovery.java
+3 -3
View File
@@ -259,9 +259,9 @@ public class NodeDiscovery {
* bitwise OR's the two bit sets or put another way, merges the source bitset into the
* destination bitset and returns the destination
*
* @param source
* @param destination
* @return
* @param source to or
* @param destination to or
* @return result.
*/
private DirectBitSet orBitSets(@NotNull final DirectBitSet source,
@NotNull final DirectBitSet destination) {
src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java
+14 -12
View File
@@ -621,7 +621,7 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
readLock();
try {
long keySize = metaKeyWriter.size(keyWriter, key);
MultiStoreBytes entry = tmpBytes;
MultiStoreBytes entry = acquireTmpBytes();
long offset = searchKey(keyWriter, metaKeyWriter, key, keySize, hash2, entry,
hashLookupLiveOnly);
if (offset >= 0) {
@@ -816,7 +816,7 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
hashLookup.startSearch(hash2);
for (long pos; (pos = hashLookup.nextPos()) >= 0L; ) {
long offset = offsetFromPos(pos);
NativeBytes entry = entry(offset);
MultiStoreBytes entry = entry(offset);
if (!keyEquals(keyWriter, metaKeyWriter, key, keySize, entry))
continue;
// key is found
@@ -982,7 +982,7 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
multiMap.startSearch(hash2);
for (long pos; (pos = multiMap.nextPos()) >= 0L; ) {
long offset = offsetFromPos(pos);
NativeBytes entry = entry(offset);
MultiStoreBytes entry = entry(offset);
if (!keyEquals(keyWriter, metaKeyWriter, key, keySize, entry))
continue;
// key is found
@@ -1049,7 +1049,7 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
hashLookupLiveOnly.startSearch(hash2);
for (long pos; (pos = hashLookupLiveOnly.nextPos()) >= 0L; ) {
long offset = offsetFromPos(pos);
NativeBytes entry = entry(offset);
MultiStoreBytes entry = entry(offset);
if (!keyEquals(keyWriter, metaKeyWriter, key, keySize, entry))
continue;
// key is found
@@ -1109,19 +1109,21 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
@Override
public Entry<K, V> getEntry(long pos) {
bytes.position(offsetFromPos(pos) + metaDataBytes);
MultiStoreBytes entry = acquireTmpBytes();
long offset = offsetFromPos(pos) + metaDataBytes;
entry.setBytesOffset(bytes, offset);
long keySize = keySizeMarshaller.readSize(bytes);
long keySize = keySizeMarshaller.readSize(entry);
ThreadLocalCopies copies = keyReaderProvider.getCopies(null);
K key = keyReaderProvider.get(copies, originalKeyReader).read(bytes, keySize);
K key = keyReaderProvider.get(copies, originalKeyReader).read(entry, keySize);
long timestamp = bytes.readLong();
bytes.skip(2L); // identifier and isDeleted flag
long timestamp = entry.readLong();
entry.skip(2L); // identifier and isDeleted flag
long valueSize = valueSizeMarshaller.readSize(bytes);
alignment.alignPositionAddr(bytes);
long valueSize = valueSizeMarshaller.readSize(entry);
alignment.alignPositionAddr(entry);
copies = valueReaderProvider.getCopies(copies);
V value = valueReaderProvider.get(copies, originalValueReader).read(bytes, valueSize);
V value = valueReaderProvider.get(copies, originalValueReader).read(entry, valueSize);
return new TimestampTrackingEntry(key, value, timestamp);
}
src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java
+1 -1
View File
@@ -790,7 +790,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
private void writeSizeAt(long locationOfSize) {
long size = bytes.position() - locationOfSize;
LOG.info("writing size = " + size);
// LOG.info("writing size = " + size);
bytes.writeInt(locationOfSize, (int) size - SIZE_OF_SIZE);
}
src/main/java/net/openhft/chronicle/map/TcpReplicator.java
+2 -2
View File
@@ -1270,14 +1270,14 @@ class StatelessServerConnector<K, V> {
@NotNull final ByteBufferBytes reader) {
LOG.info("processStatelessEvent - writer position=" + reader.position() +
/* LOG.info("processStatelessEvent - writer position=" + reader.position() +
", " +
"limit=" + reader.limit() + ",capacity=" + reader.capacity());
LOG.info("StatelessServerConnector- reader position=" + writer.position() +
", " +
"limit=" + writer.limit() + ",capacity=" + writer.capacity());
"limit=" + writer.limit() + ",capacity=" + writer.capacity());*/
final StatelessChronicleMap.EventId event = StatelessChronicleMap.EventId.values()[eventId];
src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java
+34 -28
View File
@@ -523,6 +523,14 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
return valueSize;
}
static final ThreadLocal<MultiStoreBytes> tmpBytes = new ThreadLocal<>();
protected MultiStoreBytes acquireTmpBytes() {
MultiStoreBytes b = tmpBytes.get();
if (b == null)
tmpBytes.set(b = new MultiStoreBytes());
return b;
}
// these methods should be package local, not public or private.
class Segment implements SharedSegment {
/*
@@ -537,7 +545,6 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
static final long SIZE_OFFSET = LOCK_OFFSET + 8L; // 32-bit
final NativeBytes bytes;
final MultiStoreBytes tmpBytes = new MultiStoreBytes();
final long entriesOffset;
private final int index;
private final SingleThreadedDirectBitSet freeList;
@@ -555,7 +562,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
long start = bytes.startAddr() + SEGMENT_HEADER;
createHashLookups(start);
start += align64(sizeOfMultiMap() + sizeOfMultiMapBitSet()) * multiMapsPerSegment();
final NativeBytes bsBytes = new NativeBytes(tmpBytes.objectSerializer(),
final NativeBytes bsBytes = new NativeBytes(ms.objectSerializer(),
start, start + align8(align8(entriesPerSegment) / 8L), null);
freeList = new SingleThreadedDirectBitSet(bsBytes);
start += numberOfBitSets() * sizeOfBitSets();
@@ -624,7 +631,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
public void readLock() throws IllegalStateException {
while (true) {
final boolean success = bytes.tryLockNanosLong(LOCK_OFFSET, lockTimeOutNS);
final boolean success = bytes.tryRWReadLock(LOCK_OFFSET, lockTimeOutNS);
if (success) return;
if (currentThread().isInterrupted()) {
throw new IllegalStateException(new InterruptedException("Unable to obtain lock, interrupted"));
@@ -637,7 +644,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
public void writeLock() throws IllegalStateException {
while (true) {
final boolean success = bytes.tryLockNanosLong(LOCK_OFFSET, lockTimeOutNS);
final boolean success = bytes.tryRWWriteLock(LOCK_OFFSET, lockTimeOutNS);
if (success) return;
if (currentThread().isInterrupted()) {
throw new IllegalStateException(new InterruptedException("Unable to obtain lock, interrupted"));
@@ -650,7 +657,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
public void readUnlock() {
try {
bytes.unlockLong(LOCK_OFFSET);
bytes.unlockRWReadLock(LOCK_OFFSET);
} catch (IllegalMonitorStateException e) {
errorListener.errorOnUnlock(e);
}
@@ -658,7 +665,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
public void writeUnlock() {
try {
bytes.unlockLong(LOCK_OFFSET);
bytes.unlockRWWriteLock(LOCK_OFFSET);
} catch (IllegalMonitorStateException e) {
errorListener.errorOnUnlock(e);
}
@@ -675,15 +682,12 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
public MultiStoreBytes entry(long offset) {
return reuse(tmpBytes, offset);
return reuse(acquireTmpBytes(), offset);
}
private MultiStoreBytes reuse(MultiStoreBytes entry, long offset) {
offset += metaDataBytes;
entry.storePositionAndSize(bytes, offset,
// "Infinity". Limit not used when treating entries as
// possibly oversized
bytes.limit() - offset);
entry.setBytesOffset(bytes, offset);
return entry;
}
@@ -728,7 +732,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
readLock();
try {
long keySize = metaKeyInterop.size(keyInterop, key);
MultiStoreBytes entry = tmpBytes;
MultiStoreBytes entry = acquireTmpBytes();
long offset = searchKey(keyInterop, metaKeyInterop, key, keySize, hash2, entry,
hashLookup);
if (offset >= 0L) {
@@ -776,7 +780,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
}
V onKeyPresentOnAcquire(ThreadLocalCopies copies, K key, V usingValue, long offset,
NativeBytes entry) {
MultiStoreBytes entry) {
V v = readValue(copies, entry, usingValue);
notifyGet(offset, key, v);
return v;
@@ -817,7 +821,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
hashLookup.startSearch(hash2);
for (long pos; (pos = hashLookup.nextPos()) >= 0L; ) {
long offset = offsetFromPos(pos);
NativeBytes entry = entry(offset);
MultiStoreBytes entry = entry(offset);
if (!keyEquals(keyInterop, metaKeyInterop, key, keySize, entry))
continue;
// key is found
@@ -840,7 +844,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
}
}
V replaceValueOnPut(ThreadLocalCopies copies, K key, V value, NativeBytes entry, long pos,
V replaceValueOnPut(ThreadLocalCopies copies, K key, V value, MultiStoreBytes entry, long pos,
long offset, boolean readPrevValue, MultiMap searchedHashLookup) {
long valueSizePos = entry.position();
long valueSize = readValueSize(entry);
@@ -978,15 +982,14 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
nextPosToSearchFrom = fromPos;
}
V readValue(ThreadLocalCopies copies, NativeBytes entry, V value) {
V readValue(ThreadLocalCopies copies, MultiStoreBytes entry, V value) {
return readValue(copies, entry, value, readValueSize(entry));
}
V readValue(ThreadLocalCopies copies, NativeBytes entry, V value, long valueSize) {
V readValue(ThreadLocalCopies copies, MultiStoreBytes entry, V value, long valueSize) {
copies = valueReaderProvider.getCopies(copies);
BytesReader<V> valueReader = valueReaderProvider.get(copies, originalValueReader);
bytes.positionAddr(entry.positionAddr());
return valueReader.read(bytes, valueSize, value);
return valueReader.read(entry, valueSize, value);
}
boolean keyEquals(KI keyInterop, MKI metaKeyInterop, K key,
@@ -1013,7 +1016,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
hashLookup.startSearch(hash2);
for (long pos; (pos = hashLookup.nextPos()) >= 0L; ) {
long offset = offsetFromPos(pos);
NativeBytes entry = entry(offset);
MultiStoreBytes entry = entry(offset);
if (!keyEquals(keyInterop, metaKeyInterop, key, keySize, entry))
continue;
// key is found
@@ -1074,7 +1077,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
hashLookup.startSearch(hash2);
for (long pos; (pos = hashLookup.nextPos()) >= 0L; ) {
long offset = offsetFromPos(pos);
NativeBytes entry = entry(offset);
MultiStoreBytes entry = entry(offset);
if (!keyEquals(keyInterop, metaKeyInterop, key, keySize, entry))
continue;
// key is found
@@ -1090,7 +1093,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
}
V onKeyPresentOnReplace(ThreadLocalCopies copies, K key, V expectedValue, V newValue,
long pos, long offset, NativeBytes entry,
long pos, long offset, MultiStoreBytes entry,
MultiMap searchedHashLookup) {
long valueSizePos = entry.position();
long valueSize = readValueSize(entry);
@@ -1111,24 +1114,27 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
void notifyPut(long offset, boolean added, K key, V value, final long pos) {
if (eventListener() != MapEventListeners.NOP) {
tmpBytes.storePositionAndSize(bytes, offset, entrySize);
eventListener().onPut(VanillaChronicleMap.this, tmpBytes, metaDataBytes,
MultiStoreBytes b = acquireTmpBytes();
b.storePositionAndSize(bytes, offset, entrySize);
eventListener().onPut(VanillaChronicleMap.this, b, metaDataBytes,
added, key, value, pos, this);
}
}
void notifyGet(long offset, K key, V value) {
if (eventListener() != MapEventListeners.NOP) {
tmpBytes.storePositionAndSize(bytes, offset, entrySize);
eventListener().onGetFound(VanillaChronicleMap.this, tmpBytes, metaDataBytes,
MultiStoreBytes b = acquireTmpBytes();
b.storePositionAndSize(bytes, offset, entrySize);
eventListener().onGetFound(VanillaChronicleMap.this, b, metaDataBytes,
key, value);
}
}
void notifyRemoved(long offset, K key, V value, final long pos) {
if (eventListener() != MapEventListeners.NOP) {
tmpBytes.storePositionAndSize(bytes, offset, entrySize);
eventListener().onRemove(VanillaChronicleMap.this, tmpBytes, metaDataBytes,
MultiStoreBytes b = acquireTmpBytes();
b.storePositionAndSize(bytes, offset, entrySize);
eventListener().onRemove(VanillaChronicleMap.this, b, metaDataBytes,
key, value, pos, this);
}
}