readValue, boolean resultUnused,
byte identifier, long timestamp, boolean remote) {
segmentStateNotNullImpliesCopiesNotNull(copies, segmentState);
SegmentState localSegmentState = segmentState;
if (localSegmentState == null) {
copies = SegmentState.getCopies(copies); // copies is not null now
localSegmentState = SegmentState.get(copies);
}
try {
MultiMap hashLookup = hashLookup();
hashLookup.startSearch(hash2);
MultiStoreBytes entry = localSegmentState.tmpBytes;
for (long pos; (pos = hashLookup.nextPos()) >= 0L; ) {
long offset = offsetFromPos(pos);
reuse(entry, offset);
if (!keyEquals(keyInterop, metaKeyInterop, key, keySize, entry))
continue;
// key is found
entry.skip(keySize);
final long timeStampPosAddr = entry.positionAddr();
if (shouldIgnore(entry, timestamp, identifier)) {
// the following assert should be enabled, but TimeBasedReplicationTest
// intentionally violates the explained invariant, => assertion fails.
// todo do something with this
// we should ignore only external remote updates
// which don't use put and remove results
// assert resultUnused;
return null;
}
boolean isDeleted = entry.readBoolean();
if (replaceIfPresent || isDeleted) {
entry.positionAddr(timeStampPosAddr);
entry.writeLong(timestamp);
entry.writeByte(identifier);
// deleted flag
entry.writeBoolean(false);
RV prevValue = replaceValueAndNotifyPut(copies, localSegmentState,
key, keySize, toKey,
getValueInterops, value, toValue,
entry, pos, offset, hashLookup, readValue,
resultUnused, isDeleted, remote);
// 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();
hashLookup.putPosition(localSegmentState.pos);
}
if (resultUnused)
return null;
return isDeleted ? readValue.readNull() : prevValue;
} else {
long valueSize = readValueSize(entry);
return resultUnused ? null :
readValue.readValue(copies, entry, null, valueSize);
}
}
// key is not found
VBI valueInterop = getValueInterops.getValueInterop(copies);
MVBI metaValueInterop =
getValueInterops.getMetaValueInterop(copies, valueInterop, value);
long valueSize = metaValueInterop.size(valueInterop, value);
putEntry(localSegmentState, metaKeyInterop, keyInterop, key, keySize,
metaValueInterop, valueInterop, value, entry, false);
entry.position(localSegmentState.valueSizePos - ADDITIONAL_ENTRY_BYTES);
entry.writeLong(timestamp);
entry.writeByte(identifier);
entry.writeBoolean(false);
// put callbacks
onPutMaybeRemote(localSegmentState.pos, remote);
if (bytesEventListener != null)
bytesEventListener.onPut(entry, 0L, metaDataBytes,
localSegmentState.valueSizePos, true);
if (eventListener != null)
eventListener.onPut(toKey.toInstance(copies, key, keySize),
toValue.toInstance(copies, value, valueSize), null);
return resultUnused ? null : readValue.readNull();
} finally {
if (segmentState == null && localSegmentState != null)
localSegmentState.close();
}
}
/**
* 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.
*
* We can reject put() and removes() when comparing times stamps with remote systems
*
* @param entry the maps entry
* @param timestamp the time the entry was created or updated
* @param identifier the unique identifier relating to this map
* @return true if the entry should not be processed
*/
private boolean shouldIgnore(@NotNull final NativeBytes entry, final long timestamp,
final byte identifier) {
final long lastModifiedTimeStamp = entry.readLong();
// if the readTimeStamp is newer then we'll reject this put()
// or they are the same and have a larger id
if (lastModifiedTimeStamp < timestamp) {
entry.skip(1); // skip the byte used for the identifier
return false;
}
if (lastModifiedTimeStamp > timestamp)
return true;
// check the identifier
return entry.readByte() > identifier;
}
@Override
void manageReplicationBytes(Bytes entry, boolean writeDefaultInitialReplicationValues,
boolean remove) {
if (!writeDefaultInitialReplicationValues) {
entry.skip(ADDITIONAL_ENTRY_BYTES);
} else {
entry.writeLong(currentTime());
entry.writeByte(localIdentifier);
entry.writeBoolean(remove);
}
}
@Override
,
RV, VB extends RV, VBI, MVBI extends MetaBytesInterop>
Object remove(@Nullable ThreadLocalCopies copies, @Nullable SegmentState segmentState,
MKBI metaKeyInterop, KBI keyInterop, KB key, long keySize,
InstanceOrBytesToInstance toKey,
GetValueInterops getValueInterops, VB expectedValue,
InstanceOrBytesToInstance toValue,
long hash2, ReadValue readValue, boolean resultUnused) {
writeLock();
try {
return removeWithoutLock(copies, segmentState,
metaKeyInterop, keyInterop, key, keySize, toKey,
getValueInterops, expectedValue, toValue,
hash2, readValue, resultUnused, currentTime(), localIdentifier, false,
expectedValue != null);
} finally {
if (segmentState != null)
segmentState.close();
writeUnlock();
}
}
@Override
boolean isDeleted(Bytes entry, long keySize) {
return entry.readBoolean(entry.position() + keySize + ADDITIONAL_ENTRY_BYTES - 1L);
}
@Override
,
RV, VB extends RV, VBI, MVBI extends MetaBytesInterop>
Object removeWithoutLock(
@Nullable ThreadLocalCopies copies, @Nullable SegmentState segmentState,
MKBI metaKeyInterop, KBI keyInterop, KB key, long keySize,
InstanceOrBytesToInstance toKey,
GetValueInterops getValueInterops, VB expectedValue,
InstanceOrBytesToInstance toValue,
long hash2, ReadValue readValue, boolean resultUnused) {
return removeWithoutLock(copies, segmentState, metaKeyInterop, keyInterop, key, keySize,
toKey, getValueInterops, expectedValue, toValue, hash2, readValue, resultUnused,
currentTime(), localIdentifier, false, expectedValue != null);
}
/**
* - if expectedValue is not null, returns Boolean.TRUE (removed) or Boolean.FALSE (entry
* not found), regardless the expectedValue object is Bytes instance (RPC call) or the value
* instance - if expectedValue is null: - if resultUnused is false, null or removed value is
* returned - if resultUnused is true, null is always returned
*/
,
RV, VB extends RV, VBI, MVBI extends MetaBytesInterop>
Object removeWithoutLock(
@Nullable ThreadLocalCopies copies, final @Nullable SegmentState segmentState,
MKBI metaKeyInterop, KBI keyInterop, KB key, long keySize,
InstanceOrBytesToInstance toKey,
GetValueInterops getValueInterops, VB expectedValue,
InstanceOrBytesToInstance toValue,
long hash2, ReadValue readValue, boolean resultUnused,
long timestamp, byte identifier, boolean remote, boolean booleanResult) {
segmentStateNotNullImpliesCopiesNotNull(copies, segmentState);
assert identifier > 0;
expectedValueNotNullImpliesBooleanResult(expectedValue, booleanResult);
SegmentState localSegmentState = segmentState;
try {
MultiMap hashLookup = hashLookup();
hashLookup.startSearch(hash2);
MultiStoreBytes entry = null;
returnNothing:
{
for (long pos; (pos = hashLookup.nextPos()) >= 0L; ) {
long offset = offsetFromPos(pos);
if (entry == null) {
if (localSegmentState == null) {
copies = SegmentState.getCopies(copies); // copies is not null now
localSegmentState = SegmentState.get(copies);
}
entry = localSegmentState.tmpBytes;
}
reuse(entry, offset);
if (!keyEquals(keyInterop, metaKeyInterop, key, keySize, entry))
continue;
// key is found
entry.skip(keySize);
long timestampPos = entry.position();
if (shouldIgnore(entry, timestamp, identifier)) {
// the following assert should be enabled, but TimeBasedReplicationTest
// intentionally violates the explained invariant, => assertion fails.
// todo do something with this
// we should ignore only remote updates
// which don't use remove, put results
// assert booleanResult || resultUnused;
return booleanResult ? Boolean.FALSE : null;
}
boolean isDeleted = entry.readBoolean();
if (isDeleted) {
if (expectedValue != null)
return Boolean.FALSE;
entry.position(timestampPos);
entry.writeLong(timestamp);
entry.writeByte(identifier);
onRemoveMaybeRemote(pos, remote);
break returnNothing;
}
long valueSizePos = entry.position();
long valueSize = readValueSize(entry);
long valuePos = entry.position();
// check the value assigned for the key is that we expect
if (expectedValue != null) {
VBI valueInterop = getValueInterops.getValueInterop(copies);
MVBI metaValueInterop = getValueInterops.getMetaValueInterop(
copies, valueInterop, expectedValue);
if (metaValueInterop.size(valueInterop, expectedValue) != valueSize)
return Boolean.FALSE;
if (!metaValueInterop.startsWith(valueInterop, entry, expectedValue))
return Boolean.FALSE;
}
entry.position(timestampPos);
entry.writeLong(timestamp);
entry.writeByte(identifier);
entry.writeBoolean(true);
entry.position(valuePos);
return removeEntry(copies, key, keySize, toKey, toValue, readValue,
resultUnused, hashLookup, entry, pos, valueSizePos,
valueSize, remote, false, booleanResult);
}
// key is not found
if (remote) {
long entrySize = entrySize(keySize, 0);
long pos = alloc(inBlocks(entrySize));
long offset = offsetFromPos(pos);
clearMetaData(offset);
if (entry == null) {
if (localSegmentState == null) {
copies = SegmentState.getCopies(copies); // copies is not null now
localSegmentState = SegmentState.get(copies);
}
entry = localSegmentState.tmpBytes;
}
reuse(entry, offset);
keySizeMarshaller.writeSize(entry, keySize);
metaKeyInterop.write(keyInterop, entry, key);
entry.writeLong(timestamp);
entry.writeByte(identifier);
entry.writeBoolean(true);
hashLookup.putAfterFailedSearch(pos);
hashLookup.removePosition(pos);
}
}
// return nothing
if (booleanResult) {
return Boolean.FALSE;
} else {
return resultUnused ? null : readValue.readNull();
}
} finally {
if (segmentState == null && localSegmentState != null)
localSegmentState.close();
}
}
,
RV, VB extends RV, VBI, MVBI extends MetaBytesInterop>
Object replace(
@Nullable ThreadLocalCopies copies, @Nullable SegmentState segmentState,
MKBI metaKeyInterop, KBI keyInterop, KB key, long keySize,
InstanceOrBytesToInstance toKey,
GetValueInterops getExpectedValueInterops, VB expectedValue,
GetValueInterops getNewValueInterops, VB newValue,
ReadValue readValue, InstanceOrBytesToInstance toValue,
long hash2) {
segmentStateNotNullImpliesCopiesNotNull(copies, segmentState);
long timestamp = currentTime();
byte identifier = localIdentifier;
writeLock();
try {
MultiMap hashLookup = hashLookup();
hashLookup.startSearch(hash2);
MultiStoreBytes entry = null;
for (long pos; (pos = hashLookup.nextPos()) >= 0L; ) {
long offset = offsetFromPos(pos);
if (entry == null) {
if (segmentState == null) {
copies = SegmentState.getCopies(copies);
segmentState = SegmentState.get(copies);
}
entry = segmentState.tmpBytes;
}
reuse(entry, offset);
if (!keyEquals(keyInterop, metaKeyInterop, key, keySize, entry))
continue;
// key is found
entry.skip(keySize);
long timestampPos = entry.position();
if (shouldIgnore(entry, timestamp, identifier)) {
LOG.error("Trying to replace a value for key={} on the node with id={} " +
"at time={} (current time), but the entry is updated by node " +
"with id={} at time={}. Time is not monotonic across nodes!?",
key, identifier, timestamp, entry.readByte(entry.position() - 1),
entry.readLong(entry.position() - ADDITIONAL_ENTRY_BYTES + 1));
return readValue.readNull();
}
boolean isDeleted = entry.readBoolean();
if (isDeleted)
break;
Object result = onKeyPresentOnReplace(copies, segmentState, key, keySize, toKey,
getExpectedValueInterops, expectedValue, getNewValueInterops, newValue,
readValue, toValue, pos, offset, entry, hashLookup);
if (result != Boolean.FALSE) {
entry.position(timestampPos);
entry.writeLong(timestamp);
entry.writeByte(identifier);
}
return result;
}
// key is not found
return readValue.readNull();
} finally {
if (segmentState != null)
segmentState.close();
writeUnlock();
}
}
/**
* @param bootstrapOnlyLocalEntries if true - only entries that have been create from this
* node will be published during a bootstrap
*/
public void dirtyEntries(final long timeStamp,
final ModificationIterator.EntryModifiableCallback callback,
final boolean bootstrapOnlyLocalEntries) {
readLock(null);
ThreadLocalCopies copies = SegmentState.getCopies(null);
try (SegmentState segmentState = SegmentState.get(copies)) {
final int index = Segment.this.getIndex();
final MultiStoreBytes tmpBytes = segmentState.tmpBytes;
hashLookup().forEach(new MultiMap.EntryConsumer() {
@Override
public void accept(long hash, long pos) {
final NativeBytes entry = reuse(tmpBytes, offsetFromPos(pos));
long keySize = keySizeMarshaller.readSize(entry);
entry.skip(keySize);
final long entryTimestamp = entry.readLong();
if (entryTimestamp >= timeStamp && (!bootstrapOnlyLocalEntries ||
entry.readByte() == ReplicatedChronicleMap.this.identifier()))
callback.set(index, pos);
}
});
} finally {
readUnlock();
}
}
@Override
public Entry getEntry(@NotNull SegmentState segmentState, long pos) {
Bytes entry = reuse(segmentState.tmpBytes, offsetFromPos(pos));
long keySize = keySizeMarshaller.readSize(entry);
ThreadLocalCopies copies = keyReaderProvider.getCopies(null);
K key = keyReaderProvider.get(copies, originalKeyReader).read(entry, keySize);
long timestamp = entry.readLong();
entry.skip(2L); // identifier and isDeleted flag
long valueSize = valueSizeMarshaller.readSize(entry);
alignment.alignPositionAddr(entry);
copies = valueReaderProvider.getCopies(copies);
V value = valueReaderProvider.get(copies, originalValueReader).read(entry, valueSize);
return new TimestampTrackingEntry(key, value, timestamp);
}
@Override
boolean isDeleted(long pos) {
bytes.position(offsetFromPos(pos) + metaDataBytes);
long keySize = keySizeMarshaller.readSize(bytes);
return bytes.readBoolean(bytes.position() + keySize + ADDITIONAL_ENTRY_BYTES - 1L);
}
}
@Override
void shouldNotBeCalledFromReplicatedChronicleMap(String method) {
throw new AssertionError(method + "() method should not be called by " +
"ReplicatedChronicleMap instance");
}
class TimestampTrackingEntry extends SimpleEntry {
private static final long serialVersionUID = 0L;
transient long timestamp;
public TimestampTrackingEntry(K key, V value, long timestamp) {
super(key, value);
this.timestamp = timestamp;
}
@Override
public V setValue(V value) {
long newTimestamp = timestamp = currentTime();
put(getKey(), value, localIdentifier, newTimestamp);
return super.setValue(value);
}
}
class EntryIterator extends VanillaChronicleMap.EntryIterator {
@Override
public Entry next() {
return super.next();
}
@Override
void removePresent(VanillaChronicleMap.Segment seg, long pos) {
@SuppressWarnings("unchecked")
Segment segment = (Segment) seg;
final long offset = segment.offsetFromPos(pos);
final NativeBytes entry = segment.reuse(this.entry, offset);
final long keySize = keySizeMarshaller.readSize(entry);
long keyPosition = entry.position();
entry.skip(keySize);
long timestamp = entry.readLong();
entry.position(keyPosition);
if (timestamp > ((TimestampTrackingEntry) returnedEntry).timestamp) {
// The entry was updated after being returned from iterator.next()
// Check that it is still the entry with the same key
K key = returnedEntry.getKey();
ThreadLocalCopies copies = keyInteropProvider.getCopies(null);
KI keyInterop = keyInteropProvider.get(copies, originalKeyInterop);
copies = metaKeyInteropProvider.getCopies(copies);
MKI metaKeyInterop =
metaKeyInteropProvider.get(copies, originalMetaKeyInterop, keyInterop, key);
long returnedKeySize = metaKeyInterop.size(keyInterop, key);
if (returnedKeySize != keySize ||
!metaKeyInterop.startsWith(keyInterop, entry, key)) {
// The case:
// 1. iterator.next() - thread 1
// 2. map.put() which cause relocation of the key, returned above - thread 2
// OR map.remove() which remove this key - thread 2
// 3. map.put() which place a new key on the `pos` in current segment - thread 3
// 4. iterator.remove() - thread 1
ReplicatedChronicleMap.this.remove(key);
return;
}
}
removePresent(segment, pos, entry, keySize, 0L, false);
}
}
class EntrySet extends VanillaChronicleMap.EntrySet {
@NotNull
@Override
public Iterator> iterator() {
return new EntryIterator();
}
}
/**
* Once a change occurs to a map, map replication requires that these changes are picked up
* by another thread, this class provides an iterator like interface to poll for such changes.
*
In most cases the thread that adds data to the node is unlikely to be the same thread
* that replicates the data over to the other nodes, so data will have to be marshaled between
* the main thread storing data to the map, and the thread running the replication.
One
* way to perform this marshalling, would be to pipe the data into a queue. However, This class
* takes another approach. It uses a bit set, and marks bits which correspond to the indexes of
* the entries that have changed. It then provides an iterator like interface to poll for such
* changes.
*
* @author Rob Austin.
*/
class ModificationIterator implements Replica.ModificationIterator {
private final ModificationNotifier modificationNotifier;
private final ATSDirectBitSet changes;
private final int segmentIndexShift;
private final long posMask;
private final EntryModifiableCallback entryModifiableCallback =
new EntryModifiableCallback();
// records the current position of the cursor in the bitset
private long position = -1L;
// todo get rid of this
private final MultiStoreBytes tmpBytes = new MultiStoreBytes();
/**
* @param bytes the back the bitset, used to mark which entries have changed
* @param modificationNotifier called when ever there is a change applied
*/
public ModificationIterator(@NotNull final Bytes bytes,
@NotNull final ModificationNotifier modificationNotifier) {
this.modificationNotifier = modificationNotifier;
long bitsPerSegment = bitsPerSegmentInModIterBitSet();
segmentIndexShift = Long.numberOfTrailingZeros(bitsPerSegment);
posMask = bitsPerSegment - 1L;
changes = new ATSDirectBitSet(bytes);
}
/**
* used to merge multiple segments and positions into a single index used by the bit map
*
* @param segmentIndex the index of the maps segment
* @param pos the position within this {@code segmentIndex}
* @return and index the has combined the {@code segmentIndex} and {@code pos} into a
* single value
*/
private long combine(int segmentIndex, long pos) {
return (((long) segmentIndex) << segmentIndexShift) | pos;
}
public void onPut(long pos, SharedSegment segment) {
changes.set(combine(segment.getIndex(), pos));
modificationNotifier.onChange();
}
public void onRemove(long pos, SharedSegment segment) {
changes.set(combine(segment.getIndex(), pos));
modificationNotifier.onChange();
}
/**
* Ensures that garbage in the old entry's location won't be broadcast as changed entry.
*/
void onRelocation(long pos, SharedSegment segment) {
changes.clear(combine(segment.getIndex(), pos));
}
/**
* you can continue to poll hasNext() until data becomes available. If are are in the middle
* of processing an entry via {@code nextEntry}, hasNext will return true until the bit is
* cleared
*
* @return true if there is an entry
*/
@Override
public boolean hasNext() {
final long position = this.position;
return changes.nextSetBit(position == NOT_FOUND ? 0L : position) != NOT_FOUND ||
(position > 0L && changes.nextSetBit(0L) != NOT_FOUND);
}
/**
* @param entryCallback call this to get an entry, this class will take care of the locking
* @return true if an entry was processed
*/
@Override
public boolean nextEntry(@NotNull final EntryCallback entryCallback, final int chronicleId) {
long position = this.position;
while (true) {
long oldPosition = position;
position = changes.nextSetBit(oldPosition + 1L);
if (position == NOT_FOUND) {
if (oldPosition == NOT_FOUND) {
this.position = NOT_FOUND;
return false;
}
continue;
}
this.position = position;
final VanillaChronicleMap.Segment segment =
segment((int) (position >>> segmentIndexShift));
segment.readLock(null);
try {
if (changes.clearIfSet(position)) {
entryCallback.onBeforeEntry();
final long segmentPos = position & posMask;
final NativeBytes entry =
segment.reuse(tmpBytes, segment.offsetFromPos(segmentPos));
// if the entry should be ignored, we'll move the next entry
final boolean success = entryCallback.onEntry(entry, chronicleId);
entryCallback.onAfterEntry();
if (success) {
return true;
}
}
// if the position was already cleared by another thread
// while we were trying to obtain segment lock (for example, in onRelocation()),
// go to pick up next (next iteration in while (true) loop)
} finally {
segment.readUnlock();
}
}
}
@Override
public void dirtyEntries(long fromTimeStamp) {
// iterate over all the segments and mark bit in the modification iterator
// that correspond to entries with an older timestamp
for (final Segment segment : (Segment[]) segments) {
segment.dirtyEntries(fromTimeStamp, entryModifiableCallback,
bootstrapOnlyLocalEntries);
}
}
/**
* details about when a modification to an entry was made
*/
class EntryModifiableCallback {
/**
* set the bit related to {@code segment} and {@code pos}
*
* @param segmentIndex the segment relating to the bit to set
* @param pos the position relating to the bit to set
*/
public void set(int segmentIndex, long pos) {
final long combine = combine(segmentIndex, pos);
changes.set(combine);
}
}
}
}