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

Safer unlocking on eventListener calls

Browse Source
This commit is contained in:
leventov
2015-12-16 17:33:20 +03:00
parent 93ed26e983
commit 0d716996ea
2 changed files with 49 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java
+27 -34
View File
@@ -957,6 +957,8 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
if (eventListener != null) {
V valueInstance = toValue.toInstance(copies, v, valueSize);
writeUnlock();
// TODO unlocking is dangerous, because this method is a part of
// acquireUsingLocked()
try {
eventListener.onPut(keyInstance, valueInstance, null, false, true,
true, localIdentifier, replacedIdentifier, timestamp,
@@ -1159,6 +1161,8 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
updateResult = isDeleted ? UpdateResult.INSERT : UpdateResult.UPDATE;
}
replaceValueDeletedCallback(segmentState, hashLookup, pos, isDeleted);
// put callbacks
boolean hasValueChanged = updateResult != UpdateResult.UNCHANGED;
onPutMaybeRemote(segmentState.pos, false);
@@ -1184,21 +1188,6 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
}
}
// for DRY (reusing replaceValueAndNotifyPut() method),
// size is updated AFTER callbacks are called.
// however this shouldn't be an issue because exclusive segment lock
// is still held
if (isDeleted) {
incrementSize();
// if they are NOT equal, it means the entry was relocated in putValue(),
// hence position is already set
if (pos == segmentState.pos) {
hashLookup.putPosition(segmentState.pos);
} else {
assert hashLookup.getPositions().isSet(segmentState.pos);
}
}
return updateResult;
}
// key is not found
@@ -1315,20 +1304,7 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
entry, pos, hashLookup, readValue,
resultUnused, isDeleted, remote,
replacedIdentifier, replacedTimestamp);
// for DRY (reusing replaceValueAndNotifyPut() method),
// size is updated AFTER callbacks are called.
// however this shouldn't be an issue because exclusive segment lock
// is still held
if (isDeleted) {
incrementSize();
// if they are NOT equal, it means the entry was relocated in putValue(),
// hence position is already set
if (pos == segmentState.pos) {
hashLookup.putPosition(segmentState.pos);
} else {
assert hashLookup.getPositions().isSet(segmentState.pos);
}
}
if (resultUnused)
return null;
return isDeleted ? readValue.readNull() : prevValue;
@@ -1379,6 +1355,21 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
return resultUnused ? null : readValue.readNull();
}
@Override
void replaceValueDeletedCallback(
SegmentState segmentState, MultiMap hashLookup, long pos, boolean isDeleted) {
if (isDeleted) {
incrementSize();
// if they are NOT equal, it means the entry was relocated in putValue(),
// hence position is already set
if (pos == segmentState.pos) {
hashLookup.putPosition(segmentState.pos);
} else {
assert hashLookup.getPositions().isSet(segmentState.pos);
}
}
}
/**
* Used only with replication, its sometimes possible to receive an old ( or stale update )
* from a remote map. This method is used to determine if we should ignore such updates.
@@ -1643,11 +1634,6 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
getExpectedValueInterops, expectedValue, getNewValueInterops, newValue,
readValue, toValue, pos, entry, hashLookup,
replacedIdentifier, replacedTimestamp);
if (result != Boolean.FALSE) {
entry.position(timestampPos);
entry.writeLong(timestamp);
entry.writeByte(identifier);
}
return result;
}
// key is not found
@@ -1658,6 +1644,13 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
}
}
@Override
void updateReplicationBytesOnKeyPresentOnReplace(
Bytes entry, long timestampPos, long timestamp, byte identifier) {
entry.position(timestampPos);
entry.writeLong(timestamp);
entry.writeByte(identifier);
}
public void dirtyEntries(final long timeStamp,
final EntryModifiableCallback callback,
src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java
+22 -5
View File
@@ -1976,11 +1976,6 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
"replicated map");
}
/* @Override
public void writeLock() {
writeLock();
}*/
final MultiStoreBytes reuse(MultiStoreBytes entry, long offset) {
entry.setBytesOffset(bytes, offset);
entry.position(metaDataBytes);
@@ -2127,6 +2122,8 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
final K key1 = toKey.toInstance(copies, key, keySize);
final V newValue = toValue.toInstance(copies, v, valueSize);
writeUnlock();
// TODO unlocking here is dangerous, because this method is a part of
// acquireUsingLocked(), after which lock assumed to be valid
try {
eventListener.onPut(key1,
newValue, null, false, true, true,
@@ -2431,6 +2428,8 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
sizeOfEverythingBeforeValue);
}
replaceValueDeletedCallback(segmentState, searchedHashLookup, pos, entryIsDeleted);
// put callbacks
onPutMaybeRemote(segmentState.pos, remote);
if (bytesEventListener != null)
@@ -2457,6 +2456,13 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
return resultUnused ? null : prevValue;
}
void replaceValueDeletedCallback(
SegmentState segmentState, MultiMap hashLookup, long pos, boolean isDeleted) {
if (isDeleted)
throw new AssertionError();
// do nothing
}
final void onPutMaybeRemote(long pos, boolean remote) {
if (remote) {
onRemotePut(this, pos);
@@ -2701,6 +2707,8 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
final K key1 = toKey.toInstance(copies, key, keySize);
writeUnlock();
// TODO this is called from WriteContext.removeEntry(), losing exclusive lock
// could be dangerous
try {
eventListener.onRemove(key1,
removedValueForEventListener, remote,
@@ -2852,6 +2860,10 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
sizeOfEverythingBeforeValue);
}
updateReplicationBytesOnKeyPresentOnReplace(entry,
valueSizePos - ReplicatedChronicleMap.ADDITIONAL_ENTRY_BYTES,
segmentState.timestamp, segmentState.identifier);
// put callbacks
onPut(this, segmentState.pos);
if (bytesEventListener != null) {
@@ -2882,6 +2894,11 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
return expectedValue == null ? prevValue : Boolean.TRUE;
}
void updateReplicationBytesOnKeyPresentOnReplace(
Bytes entry, long timestampPos, long timestamp, byte identifier) {
// do nothing
}
/**
* 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.