metaValueWriter = metaValueWriterProvider.get(
copies, originalMetaValueWriter, valueWriter, defaultValue);
offset = putEntry(metaKeyInterop, keyInterop, key, keySize, hash2,
localIdentifier, timestamp, hashLookupLiveOnly,
metaValueWriter, valueWriter, defaultValue);
} else {
offset = putEntry(metaKeyInterop, keyInterop, key, keySize, hash2,
localIdentifier, timestamp, hashLookupLiveOnly,
prepareValueBytesAsWriter, null, key);
}
return onKeyAbsentOnAcquire(copies, key, keySize, usingValue, offset);
}
} finally {
if (create)
writeUnlock();
else
readUnlock();
}
}
@Override
void skipReplicationBytes(Bytes bytes) {
bytes.skip(ADDITIONAL_ENTRY_BYTES);
}
/**
* called from a remote node as part of replication
*/
private void remoteRemove(Bytes keyBytes, long hash2,
final long timestamp, final byte identifier) {
writeLock();
try {
long keySize = keyBytes.remaining();
hashLookupLiveAndDeleted.startSearch(hash2);
for (long pos; (pos = hashLookupLiveAndDeleted.nextPos()) >= 0L; ) {
long offset = offsetFromPos(pos);
NativeBytes entry = entry(offset);
if (!remoteKeyEquals(keyBytes, keySize, entry))
continue;
// key is found
entry.skip(keySize);
long timeStampPos = entry.position();
if (shouldIgnore(entry, timestamp, identifier)) {
return;
}
// skip the is deleted flag
boolean wasDeleted = entry.readBoolean();
if (!wasDeleted) {
hashLookupLiveOnly.remove(hash2, pos);
decrementSize();
}
entry.position(timeStampPos);
entry.writeLong(timestamp);
assert identifier > 0;
entry.writeByte(identifier);
// was deleted
entry.writeBoolean(true);
}
// key is not found
if (LOG.isDebugEnabled())
LOG.debug("Segment.remoteRemove() : key=" + keyBytes.toString().trim() +
" was not found");
} finally {
writeUnlock();
}
}
/**
* called from a remote node when it wishes to propagate a remove event
*/
private void remotePut(@NotNull final Bytes inBytes, long hash2,
final byte identifier, final long timestamp,
long valuePos, long valueLimit) {
writeLock();
try {
// inBytes position and limit correspond to the key
final long keySize = inBytes.remaining();
hashLookupLiveAndDeleted.startSearch(hash2);
for (long pos; (pos = hashLookupLiveAndDeleted.nextPos()) >= 0L; ) {
long offset = offsetFromPos(pos);
NativeBytes entry = entry(offset);
if (!remoteKeyEquals(inBytes, keySize, entry))
continue;
// key is found
entry.skip(keySize);
final long timeStampPos = entry.positionAddr();
if (shouldIgnore(entry, timestamp, identifier)) {
return;
}
boolean wasDeleted = entry.readBoolean();
entry.positionAddr(timeStampPos);
entry.writeLong(timestamp);
assert identifier > 0;
entry.writeByte(identifier);
// deleted flag
entry.writeBoolean(false);
long valueSizePos = entry.position();
long valueSize = readValueSize(entry);
long entryEndAddr = entry.positionAddr() + valueSize;
// write the value
inBytes.limit(valueLimit);
inBytes.position(valuePos);
putValue(pos, offset, entry, valueSizePos, entryEndAddr, null, null, inBytes,
hashLookupLiveAndDeleted);
if (wasDeleted) {
// remove() would have got rid of this so we have to add it back in
hashLookupLiveOnly.put(hash2, pos);
incrementSize();
}
return;
}
// key is not found
long valueSize = valueLimit - valuePos;
long pos = alloc(inBlocks(entrySize(keySize, valueSize)));
long offset = offsetFromPos(pos);
clearMetaData(offset);
NativeBytes entry = entry(offset);
keySizeMarshaller.writeSize(entry, keySize);
// write the key
entry.write(inBytes);
entry.writeLong(timestamp);
entry.writeByte(identifier);
entry.writeBoolean(false);
valueSizeMarshaller.writeSize(entry, valueSize);
alignment.alignPositionAddr(entry);
// write the value
inBytes.limit(valueLimit);
inBytes.position(valuePos);
entry.write(inBytes);
hashLookupLiveAndDeleted.putAfterFailedSearch(pos);
hashLookupLiveOnly.put(hash2, pos);
incrementSize();
} finally {
writeUnlock();
}
}
private boolean remoteKeyEquals(Bytes keyBytes, long keySize, Bytes entry) {
return keySize == keySizeMarshaller.readSize(entry) && entry.startsWith(keyBytes);
}
V put(ThreadLocalCopies copies, MKI metaKeyWriter, KI keyWriter,
K key, V value, long hash2, boolean replaceIfPresent,
final byte identifier, final long timestamp) {
writeLock();
try {
MultiMap hashLookup = hashLookupLiveAndDeleted;
long keySize = metaKeyWriter.size(keyWriter, key);
hashLookup.startSearch(hash2);
for (long pos; (pos = hashLookup.nextPos()) >= 0L; ) {
long offset = offsetFromPos(pos);
MultiStoreBytes entry = entry(offset);
if (!keyEquals(keyWriter, metaKeyWriter, key, keySize, entry))
continue;
// key is found
entry.skip(keySize);
final long timeStampPos = entry.positionAddr();
if (shouldIgnore(entry, timestamp, identifier))
return null;
boolean wasDeleted = entry.readBoolean();
// if wasDeleted==true then we even if replaceIfPresent==false we can treat it
// the same way as replaceIfPresent true
if (replaceIfPresent || wasDeleted) {
entry.positionAddr(timeStampPos);
entry.writeLong(timestamp);
entry.writeByte(identifier);
// deleted flag
entry.writeBoolean(false);
final V prevValue = replaceValueOnPut(copies, key, value, entry, pos,
offset, !wasDeleted && !putReturnsNull, hashLookup);
if (wasDeleted) {
// remove() would have got rid of this so we have to add it back in
hashLookupLiveOnly.put(hash2, pos);
incrementSize();
return null;
} else {
return prevValue;
}
} else {
// we wont replaceIfPresent and the entry is not deleted
return putReturnsNull ? null : readValue(copies, entry, null);
}
}
// key is not found
long offset = putEntry(copies, metaKeyWriter, keyWriter, key, keySize, hash2, value,
identifier, timestamp, hashLookup);
incrementSize();
notifyPut(offset, true, key, value, posFromOffset(offset));
return null;
} finally {
writeUnlock();
}
}
/**
* 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;
}
/**
* Puts entry. If {@code value} implements {@link Byteable} interface and {@code usingValue}
* is {@code true}, the value is backed with the bytes of this entry.
*
* @param hash2 a hash of the {@code keyBytes}. Caller was searching for the
* key in the {@code searchedHashLookup} using this hash.
* @param value the value to put
* @param identifier the identifier of the outer CHM node
* @param timestamp the timestamp when the entry was put (this could be later if
* it was a remote put) this method is called only from usual
* put or acquire
* @param searchedHashLookup the hash lookup that used to find the entry based on the key
* @return offset of the written entry in the Segment bytes
* @see VanillaChronicleMap.Segment#putEntry
*/
private long putEntry(ThreadLocalCopies copies,
MKI metaKeyInterop, KI keyInterop, K key, long keySize, long hash2,
V value,
int identifier, long timestamp, MultiMap searchedHashLookup) {
copies = valueWriterProvider.getCopies(copies);
VW valueWriter = valueWriterProvider.get(copies, originalValueWriter);
copies = metaValueWriterProvider.getCopies(copies);
MetaBytesWriter metaValueWriter = metaValueWriterProvider.get(
copies, originalMetaValueWriter, valueWriter, value);
return putEntry(metaKeyInterop, keyInterop, key, keySize, hash2,
identifier, timestamp, searchedHashLookup, metaValueWriter, valueWriter, value);
}
private long putEntry(
MKI metaKeyInterop, KI keyInterop, K key, long keySize, long hash2,
final int identifier, final long timestamp, MultiMap searchedHashLookup,
MetaBytesWriter metaElemWriter, EW elemWriter, E elem) {
long valueSize = metaElemWriter.size(elemWriter, elem);
long entrySize = entrySize(keySize, valueSize);
long pos = alloc(inBlocks(entrySize));
long offset = offsetFromPos(pos);
clearMetaData(offset);
NativeBytes entry = entry(offset);
keySizeMarshaller.writeSize(entry, keySize);
metaKeyInterop.write(keyInterop, entry, key);
entry.writeLong(timestamp);
entry.writeByte(identifier);
entry.writeBoolean(false);
valueSizeMarshaller.writeSize(entry, valueSize);
alignment.alignPositionAddr(entry);
metaElemWriter.write(elemWriter, entry, elem);
// we have to add it both to the live
if (searchedHashLookup == hashLookupLiveAndDeleted) {
hashLookupLiveAndDeleted.putAfterFailedSearch(pos);
hashLookupLiveOnly.put(hash2, pos);
} else {
hashLookupLiveOnly.putAfterFailedSearch(pos);
hashLookupLiveAndDeleted.put(hash2, pos);
}
return offset;
}
/**
* @see VanillaChronicleMap.Segment#remove
*/
V remove(ThreadLocalCopies copies, MKI metaKeyWriter, KI keyWriter,
K key, V expectedValue, long hash2,
final long timestamp, final byte identifier) {
assert identifier > 0;
writeLock();
try {
long keySize = metaKeyWriter.size(keyWriter, key);
final MultiMap multiMap = hashLookupLiveAndDeleted;
multiMap.startSearch(hash2);
for (long pos; (pos = multiMap.nextPos()) >= 0L; ) {
long offset = offsetFromPos(pos);
MultiStoreBytes entry = entry(offset);
if (!keyEquals(keyWriter, metaKeyWriter, key, keySize, entry))
continue;
// key is found
entry.skip(keySize);
long timeStampPos = entry.position();
if (shouldIgnore(entry, timestamp, identifier)) {
return null;
}
// skip the is deleted flag
final boolean wasDeleted = entry.readBoolean();
if (wasDeleted) {
// this caters for the case when the entry in not in our hashLookupLiveOnly
// map but maybe in our hashLookupLiveAndDeleted,
// so we have to send the deleted notification
entry.position(timeStampPos);
entry.writeLong(timestamp);
entry.writeByte(identifier);
// was deleted is already true
entry.skip(1);
notifyRemoved(offset, key, null, pos);
return null;
}
long valueSize = readValueSize(entry);
V valueRemoved = expectedValue != null || !removeReturnsNull
? readValue(copies, entry, null, valueSize) : null;
if (expectedValue != null && !expectedValue.equals(valueRemoved))
return null;
hashLookupLiveOnly.remove(hash2, pos);
decrementSize();
entry.position(timeStampPos);
entry.writeLong(timestamp);
entry.writeByte(identifier);
entry.writeBoolean(true);
notifyRemoved(offset, key, valueRemoved, pos);
return valueRemoved;
}
// key is not found
return null;
} finally {
writeUnlock();
}
}
@Override
MultiMap containsKeyHashLookup() {
return hashLookupLiveOnly;
}
/**
* @see VanillaChronicleMap.Segment#replace
*/
V replace(ThreadLocalCopies copies, MKI metaKeyWriter, KI keyWriter,
K key, V expectedValue, V newValue, long hash2, long timestamp) {
writeLock();
try {
long keySize = metaKeyWriter.size(keyWriter, key);
hashLookupLiveOnly.startSearch(hash2);
for (long pos; (pos = hashLookupLiveOnly.nextPos()) >= 0L; ) {
long offset = offsetFromPos(pos);
MultiStoreBytes entry = entry(offset);
if (!keyEquals(keyWriter, metaKeyWriter, key, keySize, entry))
continue;
// key is found
entry.skip(keySize);
if (shouldIgnore(entry, timestamp, localIdentifier))
return null;
// skip the is deleted flag
entry.skip(1);
return onKeyPresentOnReplace(copies, key, expectedValue, newValue, pos, offset,
entry, hashLookupLiveOnly);
}
// key is not found
return null;
} finally {
writeUnlock();
}
}
@Override
void replacePosInHashLookupOnRelocation(MultiMap searchedHashLookup,
long prevPos, long pos) {
searchedHashLookup.replacePrevPos(pos);
long hash = searchedHashLookup.getSearchHash();
MultiMap anotherLookup = searchedHashLookup == hashLookupLiveAndDeleted ?
hashLookupLiveOnly : hashLookupLiveAndDeleted;
anotherLookup.replace(hash, prevPos, pos);
}
public void dirtyEntries(final long timeStamp,
final ModificationIterator.EntryModifiableCallback callback) {
this.readLock();
try {
final int index = Segment.this.getIndex();
hashLookupLiveAndDeleted.forEach(new MultiMap.EntryConsumer() {
@Override
public void accept(long hash, long pos) {
final NativeBytes entry = entry(offsetFromPos(pos));
long keySize = keySizeMarshaller.readSize(entry);
entry.skip(keySize);
final long entryTimestamp = entry.readLong();
if (entryTimestamp >= timeStamp &&
entry.readByte() == ReplicatedChronicleMap.this.identifier())
callback.set(index, pos);
}
});
} finally {
readUnlock();
}
}
@Override
public Entry getEntry(long pos) {
MultiStoreBytes entry = acquireTmpBytes();
long offset = offsetFromPos(pos) + metaDataBytes;
entry.setBytesOffset(bytes, offset);
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
void clear() {
// we have to make sure that every calls notifies on remove,
// so that the replicators can pick it up
for (K k : keySet()) {
ReplicatedChronicleMap.this.remove(k);
}
}
}
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 = timeProvider.currentTimeMillis();
put(getKey(), value, localIdentifier, newTimestamp);
return super.setValue(value);
}
}
class EntryIterator extends VanillaChronicleMap.EntryIterator {
@Override
void removePresent(VanillaChronicleMap.Segment seg, long pos) {
@SuppressWarnings("unchecked")
Segment segment = (Segment) seg;
final long offset = segment.offsetFromPos(pos);
final NativeBytes entry = segment.entry(offset);
final long limit = entry.limit();
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 keyWriter = keyInteropProvider.get(copies, originalKeyInterop);
copies = metaKeyInteropProvider.getCopies(copies);
MKI metaKeyWriter =
metaKeyInteropProvider.get(copies, originalMetaKeyInterop, keyWriter, key);
long returnedKeySize = metaKeyWriter.size(keyWriter, key);
if (returnedKeySize != keySize ||
!metaKeyWriter.startsWith(keyWriter, 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;
}
}
final long segmentHash = segmentHash(hash(entry, keyPosition, keyPosition + keySize));
segment.hashLookupLiveOnly.remove(segmentHash, pos);
segment.decrementSize();
entry.skip(keySize);
entry.writeLong(timeProvider.currentTimeMillis());
entry.writeByte(localIdentifier);
entry.writeBoolean(true);
segment.notifyRemoved(offset, returnedEntry.getKey(), returnedEntry.getValue(), pos);
}
}
class EntrySet extends VanillaChronicleMap.EntrySet {
@NotNull
@Override
public Iterator> iterator() {
return new EntryIterator();
}
}
/**
* receive an update from the map, via the MapEventListener and delegates the changes to the
* currently active modification iterators
*/
class ModificationDelegator extends MapEventListener> {
private static final long serialVersionUID = 0L;
// the assigned modification iterators
private final ATSDirectBitSet modIterSet;
private final MapEventListener> nextListener;
private final AtomicReferenceArray modificationIterators =
new AtomicReferenceArray(127 + RESERVED_MOD_ITER);
public ModificationDelegator(@NotNull final MapEventListener> nextListener,
final Bytes bytes) {
this.nextListener = nextListener;
modIterSet = new ATSDirectBitSet(bytes);
}
public ModificationIterator acquireModificationIterator(
final short remoteIdentifier,
@NotNull final ModificationNotifier modificationNotifier) {
final ModificationIterator modificationIterator = modificationIterators.get(remoteIdentifier);
if (modificationIterator != null)
return modificationIterator;
synchronized (modificationIterators) {
final ModificationIterator modificationIterator0 = modificationIterators.get(remoteIdentifier);
if (modificationIterator0 != null)
return modificationIterator0;
final Bytes bytes = ms.bytes(startOfModificationIterators + (modIterBitSetSizeInBytes()
* remoteIdentifier),
modIterBitSetSizeInBytes());
final ModificationIterator newModificationIterator = new ModificationIterator(
bytes, modificationNotifier);
modificationIterators.set(remoteIdentifier, newModificationIterator);
modIterSet.set(remoteIdentifier);
return newModificationIterator;
}
}
@Override
public void onPut(ChronicleMap map, Bytes entry, int metaDataBytes,
boolean added, K key, V value, long pos, SharedSegment segment) {
assert ReplicatedChronicleMap.this == map :
"ModificationIterator.onPut() is called from outside of the parent map";
try {
nextListener.onPut(map, entry, metaDataBytes, added, key, value, pos, segment);
} catch (Exception e) {
LOG.error("", e);
}
for (long next = modIterSet.nextSetBit(0); next > 0; next = modIterSet.nextSetBit(next + 1)) {
try {
final ModificationIterator modificationIterator = modificationIterators.get((int) next);
modificationIterator.onPut(map, entry, metaDataBytes,
added, key, value, pos, segment);
} catch (Exception e) {
LOG.error("", e);
}
}
}
@Override
public void onRemove(ChronicleMap map, Bytes entry, int metaDataBytes,
K key, V value, long pos, SharedSegment segment) {
assert ReplicatedChronicleMap.this == map :
"ModificationIterator.onRemove() is called from outside of the parent map";
try {
nextListener.onRemove(map, entry, metaDataBytes, key, value, pos, segment);
} catch (Exception e) {
LOG.error("", e);
}
for (long next = modIterSet.nextSetBit(0); next > 0; next = modIterSet.nextSetBit(next + 1)) {
try {
final ModificationIterator modificationIterator = modificationIterators.get((int) next);
modificationIterator.onRemove(map, entry, metaDataBytes, key, value, pos, segment);
} catch (Exception e) {
LOG.error("", e);
}
}
}
@Override
void onRelocation(long pos, SharedSegment segment) {
for (long next = modIterSet.nextSetBit(0); next > 0; next = modIterSet.nextSetBit(next + 1)) {
try {
final ModificationIterator modificationIterator = modificationIterators.get((int) next);
modificationIterator.onRelocation(pos, segment);
} catch (Exception e) {
LOG.error("", e);
}
}
}
@Override
public void onGetFound(ChronicleMap map, Bytes entry, int metaDataBytes,
K key, V value) {
nextListener.onGetFound(map, entry, metaDataBytes, key, value);
}
/**
* Always throws {@code NotSerializableException} since instances of this class are not
* intended to be serializable.
*/
private void writeObject(ObjectOutputStream out) throws IOException {
throw new NotSerializableException(getClass().getCanonicalName());
}
/**
* Always throws {@code NotSerializableException} since instances of this class are not
* intended to be serializable.
*/
private void readObject(ObjectInputStream in)
throws IOException, ClassNotFoundException {
throw new NotSerializableException(getClass().getCanonicalName());
}
}
/**
* 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 extends MapEventListener>
implements Replica.ModificationIterator {
private static final long serialVersionUID = 0L;
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;
/**
* @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;
}
@Override
public void onPut(ChronicleMap map, Bytes entry, int metaDataBytes,
boolean added, K key, V value, long pos, SharedSegment segment) {
assert ReplicatedChronicleMap.this == map :
"ModificationIterator.onPut() is called from outside of the parent map";
changes.set(combine(segment.getIndex(), pos));
modificationNotifier.onChange();
}
@Override
public void onRemove(ChronicleMap map, Bytes entry, int metaDataBytes,
K key, V value, long pos, SharedSegment segment) {
assert ReplicatedChronicleMap.this == map :
"ModificationIterator.onRemove() is called from outside of the parent map";
changes.set(combine(segment.getIndex(), pos));
modificationNotifier.onChange();
}
/**
* Ensures that garbage in the old entry's location won't be broadcast as changed entry.
*/
@Override
void onRelocation(long pos, SharedSegment segment) {
changes.clear(combine(segment.getIndex(), pos));
// don't call nextListener.onRelocation(),
// because no one event listener else overrides this method.
}
/**
* 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
*/
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
*/
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 SharedSegment segment = segment((int) (position >>> segmentIndexShift));
segment.readLock();
try {
if (changes.clearIfSet(position)) {
entryCallback.onBeforeEntry();
final long segmentPos = position & posMask;
final NativeBytes entry = segment.entry(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 onReplication()),
// 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);
}
}
/**
* Always throws {@code NotSerializableException} since instances of this class are not
* intended to be serializable.
*/
private void writeObject(ObjectOutputStream out) throws IOException {
throw new NotSerializableException(getClass().getCanonicalName());
}
/**
* Always throws {@code NotSerializableException} since instances of this class are not
* intended to be serializable.
*/
private void readObject(ObjectInputStream in)
throws IOException, ClassNotFoundException {
throw new NotSerializableException(getClass().getCanonicalName());
}
/**
* 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);
}
}
}
}