diff --git a/src/main/java/net/openhft/chronicle/map/Int24Int24MultiMap.java b/src/main/java/net/openhft/chronicle/map/Int24Int24MultiMap.java index 2260b1d94..f98287d68 100644 --- a/src/main/java/net/openhft/chronicle/map/Int24Int24MultiMap.java +++ b/src/main/java/net/openhft/chronicle/map/Int24Int24MultiMap.java @@ -44,9 +44,6 @@ class Int24Int24MultiMap implements MultiMap { private final long capacityInBytes; private final Bytes bytes; - private long searchHash = -1L; - private long searchPos = -1L; - public Int24Int24MultiMap(long minCapacity) { long capacity = MultiMapFactory.multiMapCapacity(minCapacity); this.capacity = (int) capacity; @@ -208,57 +205,53 @@ class Int24Int24MultiMap implements MultiMap { // Stateful methods @Override - public void startSearch(long key) { + public void startSearch(long key, SearchState searchStateToReuse) { key = maskUnsetKey(key); - searchPos = pos(key); - searchHash = key; + searchStateToReuse.searchPos = pos(key); + searchStateToReuse.searchHash = key; } @Override - public long nextPos() { - long pos = searchPos; + public long nextPos(SearchState searchState) { + long pos = searchState.searchPos; while (true) { long entry = bytes.readInt48(pos); if (entry == UNSET_ENTRY) { - searchPos = pos; + searchState.searchPos = pos; return -1L; } pos = step(pos); - if (key(entry) == searchHash) { - searchPos = pos; + if (key(entry) == searchState.searchHash) { + searchState.searchPos = pos; return value(entry); } } } @Override - public void removePrevPos() { - long prevPos = stepBack(searchPos); + public void removePrevPos(SearchState searchState) { + long prevPos = stepBack(searchState.searchPos); long entry = bytes.readInt48(prevPos); positions.clear(value(entry)); removePos(prevPos); } @Override - public void replacePrevPos(long newValue) { + public void replacePrevPos(SearchState searchState, long newValue) { checkValueForPut(newValue); - long prevPos = stepBack(searchPos); + long prevPos = stepBack(searchState.searchPos); long oldEntry = bytes.readInt48(prevPos); long oldValue = value(oldEntry); positions.clear(oldValue); positions.set(newValue); - bytes.writeInt48(prevPos, entry(searchHash, newValue)); + bytes.writeInt48(prevPos, entry(searchState.searchHash, newValue)); } @Override - public void putAfterFailedSearch(long value) { + public void putAfterFailedSearch(SearchState searchState, long value) { checkValueForPut(value); positions.set(value); - bytes.writeInt48(searchPos, entry(searchHash, value)); - } - - public long getSearchHash() { - return searchHash; + bytes.writeInt48(searchState.searchPos, entry(searchState.searchHash, value)); } @Override diff --git a/src/main/java/net/openhft/chronicle/map/IntIntMultiMap.java b/src/main/java/net/openhft/chronicle/map/IntIntMultiMap.java index 37000c770..5d546d9cb 100644 --- a/src/main/java/net/openhft/chronicle/map/IntIntMultiMap.java +++ b/src/main/java/net/openhft/chronicle/map/IntIntMultiMap.java @@ -47,9 +47,6 @@ class IntIntMultiMap implements MultiMap { private final long capacityMask2; private final Bytes bytes; - private long searchHash = -1L; - private long searchPos = -1L; - public IntIntMultiMap(long minCapacity) { capacity = multiMapCapacity(minCapacity); capacityMask = capacity - 1L; @@ -208,57 +205,53 @@ class IntIntMultiMap implements MultiMap { // Stateful methods @Override - public void startSearch(long key) { + public void startSearch(long key, SearchState searchStateToReuse) { key = maskUnsetKey(key); - searchPos = pos(key); - searchHash = key; + searchStateToReuse.searchPos = pos(key); + searchStateToReuse.searchHash = key; } @Override - public long nextPos() { - long pos = searchPos; + public long nextPos(SearchState searchState) { + long pos = searchState.searchPos; while (true) { long entry = bytes.readLong(pos); if (entry == UNSET_ENTRY) { - searchPos = pos; + searchState.searchPos = pos; return -1L; } pos = step(pos); - if (key(entry) == searchHash) { - searchPos = pos; + if (key(entry) == searchState.searchHash) { + searchState.searchPos = pos; return value(entry); } } } @Override - public void removePrevPos() { - long prevPos = stepBack(searchPos); + public void removePrevPos(SearchState searchState) { + long prevPos = stepBack(searchState.searchPos); long entry = bytes.readLong(prevPos); positions.clear(value(entry)); removePos(prevPos); } @Override - public void replacePrevPos(long newValue) { + public void replacePrevPos(SearchState searchState, long newValue) { checkValueForPut(newValue); - long prevPos = stepBack(searchPos); + long prevPos = stepBack(searchState.searchPos); long oldEntry = bytes.readLong(prevPos); long oldValue = value(oldEntry); positions.clear(oldValue); positions.set(newValue); - bytes.writeLong(prevPos, entry(searchHash, newValue)); + bytes.writeLong(prevPos, entry(searchState.searchHash, newValue)); } @Override - public void putAfterFailedSearch(long value) { + public void putAfterFailedSearch(SearchState searchState, long value) { checkValueForPut(value); positions.set(value); - bytes.writeLong(searchPos, entry(searchHash, value)); - } - - public long getSearchHash() { - return searchHash; + bytes.writeLong(searchState.searchPos, entry(searchState.searchHash, value)); } @Override diff --git a/src/main/java/net/openhft/chronicle/map/MultiMap.java b/src/main/java/net/openhft/chronicle/map/MultiMap.java index 4275ac1c8..6f335e549 100644 --- a/src/main/java/net/openhft/chronicle/map/MultiMap.java +++ b/src/main/java/net/openhft/chronicle/map/MultiMap.java @@ -54,22 +54,20 @@ interface MultiMap { /** * Used for start a search for a given key */ - void startSearch(long key); - - long getSearchHash(); + void startSearch(long key, SearchState searchStateToReuse); /** * Used for getting the next position for a given key * * @return the next position for the last search or negative value */ - long nextPos(); + long nextPos(SearchState searchState); - void removePrevPos(); + void removePrevPos(SearchState searchState); - void replacePrevPos(long newValue); + void replacePrevPos(SearchState searchState, long newValue); - void putAfterFailedSearch(long value); + void putAfterFailedSearch(SearchState searchState, long value); void clear(); diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java index 43fb00602..066a0aab5 100644 --- a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java @@ -263,11 +263,10 @@ final class ReplicatedChronicleMap, RV> RV acquireWithoutLock( - @Nullable ThreadLocalCopies copies, final @Nullable SegmentState segmentState, + @NotNull ThreadLocalCopies copies, @NotNull SegmentState segmentState, MKBI metaKeyInterop, KBI keyInterop, KB key, long keySize, InstanceOrBytesToInstance toKey, ReadValue readValue, RV usingValue, InstanceOrBytesToInstance toValue, long hash2, boolean create, MutableLockedEntry lock) { - segmentStateNotNullImpliesCopiesNotNull(copies, segmentState); - SegmentState localSegmentState = segmentState; - try { - MultiStoreBytes entry = null; - MultiMap hashLookup = hashLookup(); - hashLookup.startSearch(hash2); - for (long pos; (pos = hashLookup.nextPos()) >= 0L; ) { - long offset = offsetFromPos(pos); - if (entry == null) { - if (localSegmentState == null) { - copies = SegmentState.getCopies(copies); - localSegmentState = SegmentState.get(copies); - } - entry = localSegmentState.tmpBytes; - } - reuse(entry, offset); - if (!keyEquals(keyInterop, metaKeyInterop, key, keySize, entry)) - continue; - // key is found - entry.skip(keySize); - entry.skip(ADDITIONAL_ENTRY_BYTES - 1L); - boolean isDeleted = entry.readBoolean(); - if (isDeleted) { - if (!create) - return readValue.readNull(); + MultiStoreBytes entry = segmentState.tmpBytes; + MultiMap hashLookup = hashLookup(); + SearchState searchState = segmentState.searchState; + hashLookup.startSearch(hash2, searchState); + for (long pos; (pos = hashLookup.nextPos(searchState)) >= 0L; ) { + long offset = offsetFromPos(pos); + reuse(entry, offset); + if (!keyEquals(keyInterop, metaKeyInterop, key, keySize, entry)) + continue; + // key is found + entry.skip(keySize); + entry.skip(ADDITIONAL_ENTRY_BYTES - 1L); + boolean isDeleted = entry.readBoolean(); + if (isDeleted) { + if (!create) + return readValue.readNull(); - long valueSizePos = entry.position(); + long valueSizePos = entry.position(); - entry.position(valueSizePos - ADDITIONAL_ENTRY_BYTES); - // todo theoretically, currentTime() call should be outside locking - // because locking might take time > 1 ms. but making current time a param - // of acquireWithoutLock would complicate code much, requiring putting - // lookupUsing() back into ReplicatedChMap, and other methods - entry.writeLong(currentTime()); - entry.writeByte(localIdentifier); - // deleted flag - entry.writeBoolean(false); + entry.position(valueSizePos - ADDITIONAL_ENTRY_BYTES); + // todo theoretically, currentTime() call should be outside locking + // because locking might take time > 1 ms. but making current time a param + // of acquireWithoutLock would complicate code much, requiring putting + // lookupUsing() back into ReplicatedChMap, and other methods + entry.writeLong(currentTime()); + entry.writeByte(localIdentifier); + // deleted flag + entry.writeBoolean(false); - long prevValueSize = readValueSize(entry); - long valueAddr = entry.positionAddr(); - long entryEndAddr = valueAddr + prevValueSize; + long prevValueSize = readValueSize(entry); + long valueAddr = entry.positionAddr(); + long entryEndAddr = valueAddr + prevValueSize; - // todo add api which doesn't require key instance - K keyInstance = toKey.toInstance(copies, key, keySize); - MetaBytesWriter metaElemWriter; - Object elemWriter; - Object elem; - if (defaultValueProvider != null) { - V defaultValue = defaultValueProvider.get(keyInstance); - elem = defaultValue; - VI valueInterop = - valueInteropProvider.get(copies, originalValueInterop); - elemWriter = valueInterop; - metaElemWriter = metaValueInteropProvider.get( - copies, originalMetaValueInterop, valueInterop, defaultValue); - } else if (prepareValueBytesAsWriter != null) { - elem = keyInstance; - elemWriter = null; - metaElemWriter = prepareValueBytesAsWriter; - } else { - throw defaultValueOrPrepareBytesShouldBeSpecified(); - } - putValue(pos, offset, entry, valueSizePos, entryEndAddr, - localSegmentState, metaElemWriter, elemWriter, elem, - metaElemWriter.size(elemWriter, elem), hashLookup); - pos = localSegmentState.pos; - - incrementSize(); - hashLookup.putPosition(pos); - - entry.position(valueSizePos); - long valueSize = readValueSize(entry); - long valuePos = entry.position(); - RV v = readValue.readValue(copies, entry, usingValue, valueSize); - - // put callbacks - onPut(this, pos); - if (bytesEventListener != null) { - bytesEventListener.onPut(entry, 0L, metaDataBytes, valuePos, true); - } - if (eventListener != null) { - V valueInstance = toValue.toInstance(copies, v, valueSize); - eventListener.onPut(keyInstance, valueInstance, null); - } - - return v; + // todo add api which doesn't require key instance + K keyInstance = toKey.toInstance(copies, key, keySize); + MetaBytesWriter metaElemWriter; + Object elemWriter; + Object elem; + if (defaultValueProvider != null) { + V defaultValue = defaultValueProvider.get(keyInstance); + elem = defaultValue; + VI valueInterop = valueInteropProvider.get(copies, originalValueInterop); + elemWriter = valueInterop; + metaElemWriter = metaValueInteropProvider.get( + copies, originalMetaValueInterop, valueInterop, defaultValue); + } else if (prepareValueBytesAsWriter != null) { + elem = keyInstance; + elemWriter = null; + metaElemWriter = prepareValueBytesAsWriter; } else { - return readValueAndNotifyGet(copies, key, keySize, toKey, - readValue, usingValue, toValue, entry); + throw defaultValueOrPrepareBytesShouldBeSpecified(); } - } - if (!create) - return readValue.readNull(); - if (entry == null) { - if (localSegmentState == null) { - copies = SegmentState.getCopies(copies); // after this, copies is not null - localSegmentState = SegmentState.get(copies); + putValue(pos, offset, entry, valueSizePos, entryEndAddr, + segmentState, metaElemWriter, elemWriter, elem, + metaElemWriter.size(elemWriter, elem), hashLookup); + pos = segmentState.pos; + + incrementSize(); + hashLookup.putPosition(pos); + + entry.position(valueSizePos); + long valueSize = readValueSize(entry); + long valuePos = entry.position(); + RV v = readValue.readValue(copies, entry, usingValue, valueSize); + + // put callbacks + onPut(this, pos); + if (bytesEventListener != null) { + bytesEventListener.onPut(entry, 0L, metaDataBytes, valuePos, true); } - entry = localSegmentState.tmpBytes; + if (eventListener != null) { + V valueInstance = toValue.toInstance(copies, v, valueSize); + eventListener.onPut(keyInstance, valueInstance, null); + } + + return v; + } else { + return readValueAndNotifyGet(copies, key, keySize, toKey, + readValue, usingValue, toValue, entry); } - - - RV result = createEntryOnAcquire(copies, localSegmentState, - metaKeyInterop, keyInterop, key, keySize, toKey, - readValue, usingValue, toValue, entry); - - // notify the context that the entry was created - if (lock instanceof WriteLocked) - ((WriteLocked) lock).created(false); - - return result; - - } finally { - if (segmentState == null && localSegmentState != null) - localSegmentState.close(); } + if (!create) + return readValue.readNull(); + + RV result = createEntryOnAcquire(copies, segmentState, + metaKeyInterop, keyInterop, key, keySize, toKey, + readValue, usingValue, toValue, entry); + + // notify the context that the entry was created + if (lock instanceof WriteLocked) + ((WriteLocked) lock).created(false); + + return result; } /** @@ -879,7 +855,7 @@ final class ReplicatedChronicleMap, RV, VB extends RV, VBI, MVBI extends MetaBytesInterop> - RV put(@Nullable ThreadLocalCopies copies, final @Nullable SegmentState segmentState, + RV put(@Nullable ThreadLocalCopies copies, @Nullable SegmentState segmentState, MKBI metaKeyInterop, KBI keyInterop, KB key, long keySize, InstanceOrBytesToInstance toKey, GetValueInterops getValueInterops, VB value, @@ -887,6 +863,11 @@ final class ReplicatedChronicleMap readValue, boolean resultUnused, byte identifier, long timeStamp) { + segmentStateNotNullImpliesCopiesNotNull(copies, segmentState); + if (segmentState == null) { + copies = SegmentState.getCopies(copies); + segmentState = SegmentState.get(copies); + } writeLock(); try { return putWithoutLock(copies, segmentState, @@ -895,8 +876,7 @@ final class ReplicatedChronicleMap 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); + MultiMap hashLookup = hashLookup(); + SearchState searchState = segmentState.searchState; + hashLookup.startSearch(hash2, searchState); + MultiStoreBytes entry = segmentState.tmpBytes; + for (long pos; (pos = hashLookup.nextPos(searchState)) >= 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(); + 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 + 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); - } + // we should ignore only external remote updates + // which don't use put and remove results + // assert resultUnused; + return null; } - // 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); + boolean isDeleted = entry.readBoolean(); - return resultUnused ? null : readValue.readNull(); - } finally { - if (segmentState == null && localSegmentState != null) - localSegmentState.close(); + if (replaceIfPresent || isDeleted) { + entry.positionAddr(timeStampPosAddr); + entry.writeLong(timestamp); + entry.writeByte(identifier); + // deleted flag + entry.writeBoolean(false); + + RV prevValue = replaceValueAndNotifyPut(copies, segmentState, + 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(segmentState.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(segmentState, metaKeyInterop, keyInterop, key, keySize, + metaValueInterop, valueInterop, value, entry, false); + entry.position(segmentState.valueSizePos - ADDITIONAL_ENTRY_BYTES); + entry.writeLong(timestamp); + entry.writeByte(identifier); + entry.writeBoolean(false); + + // put callbacks + onPutMaybeRemote(segmentState.pos, remote); + if (bytesEventListener != null) + bytesEventListener.onPut(entry, 0L, metaDataBytes, + segmentState.valueSizePos, true); + if (eventListener != null) + eventListener.onPut(toKey.toInstance(copies, key, keySize), + toValue.toInstance(copies, value, valueSize), null); + + return resultUnused ? null : readValue.readNull(); } /** @@ -1067,6 +1037,11 @@ final class ReplicatedChronicleMap getValueInterops, VB expectedValue, InstanceOrBytesToInstance toValue, long hash2, ReadValue readValue, boolean resultUnused) { + segmentStateNotNullImpliesCopiesNotNull(copies, segmentState); + if (segmentState == null) { + copies = SegmentState.getCopies(copies); + segmentState = SegmentState.get(copies); + } writeLock(); try { return removeWithoutLock(copies, segmentState, @@ -1075,8 +1050,7 @@ final class ReplicatedChronicleMap, RV, VB extends RV, VBI, MVBI extends MetaBytesInterop> Object removeWithoutLock( - @Nullable ThreadLocalCopies copies, final @Nullable SegmentState segmentState, + @NotNull ThreadLocalCopies copies, @NotNull 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); + MultiMap hashLookup = hashLookup(); + SearchState searchState = segmentState.searchState; + hashLookup.startSearch(hash2, searchState); + MultiStoreBytes entry = segmentState.tmpBytes; + returnNothing: + { + for (long pos; (pos = hashLookup.nextPos(searchState)) >= 0L; ) { + long offset = offsetFromPos(pos); + 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; - } + 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); - entry.writeBoolean(true); - entry.position(valuePos); - - return removeEntry(copies, key, keySize, toKey, toValue, readValue, - resultUnused, hashLookup, entry, pos, valueSizePos, - valueSize, remote, false, booleanResult); + onRemoveMaybeRemote(pos, remote); + break returnNothing; } - // 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); + long valueSizePos = entry.position(); + long valueSize = readValueSize(entry); + long valuePos = entry.position(); - entry.writeLong(timestamp); - entry.writeByte(identifier); - entry.writeBoolean(true); - - hashLookup.putAfterFailedSearch(pos); - hashLookup.removePosition(pos); + // 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, searchState, key, keySize, toKey, toValue, readValue, + resultUnused, hashLookup, entry, pos, valueSizePos, + valueSize, remote, false, booleanResult); } - // return nothing - if (booleanResult) { - return Boolean.FALSE; - } else { - return resultUnused ? null : readValue.readNull(); + // key is not found + if (remote) { + long entrySize = entrySize(keySize, 0); + long pos = alloc(inBlocks(entrySize)); + long offset = offsetFromPos(pos); + clearMetaData(offset); + reuse(entry, offset); + + keySizeMarshaller.writeSize(entry, keySize); + metaKeyInterop.write(keyInterop, entry, key); + + entry.writeLong(timestamp); + entry.writeByte(identifier); + entry.writeBoolean(true); + + hashLookup.putAfterFailedSearch(searchState, pos); + hashLookup.removePosition(pos); } - } finally { - if (segmentState == null && localSegmentState != null) - localSegmentState.close(); + } + // return nothing + if (booleanResult) { + return Boolean.FALSE; + } else { + return resultUnused ? null : readValue.readNull(); } } @@ -1238,22 +1192,20 @@ final class ReplicatedChronicleMap readValue, InstanceOrBytesToInstance toValue, long hash2) { segmentStateNotNullImpliesCopiesNotNull(copies, segmentState); + if (segmentState == null) { + copies = SegmentState.getCopies(copies); + segmentState = SegmentState.get(copies); + } long timestamp = currentTime(); byte identifier = localIdentifier; writeLock(); try { MultiMap hashLookup = hashLookup(); - hashLookup.startSearch(hash2); - MultiStoreBytes entry = null; - for (long pos; (pos = hashLookup.nextPos()) >= 0L; ) { + SearchState searchState = segmentState.searchState; + hashLookup.startSearch(hash2, searchState); + MultiStoreBytes entry = segmentState.tmpBytes; + for (long pos; (pos = hashLookup.nextPos(searchState)) >= 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; @@ -1263,8 +1215,9 @@ final class ReplicatedChronicleMap, return segmentStateProvider.getCopies(copies); } - static SegmentState get(@NotNull ThreadLocalCopies copies) { + static @NotNull SegmentState get(@NotNull ThreadLocalCopies copies) { assertNotNull(copies); return segmentStateProvider.get(copies, originalSegmentState).get(); } @@ -1617,6 +1617,8 @@ class VanillaChronicleMap, long pos; long valueSizePos; + SearchState searchState = new SearchState(); + private SegmentState(int depth) { if (depth > (1 << 16)) throw new IllegalStateException("More than " + (1 << 16) + @@ -1909,77 +1911,61 @@ class VanillaChronicleMap, } , RV> - RV acquire(@Nullable ThreadLocalCopies copies, final @Nullable SegmentState segmentState, + RV acquire(@Nullable ThreadLocalCopies copies, @Nullable SegmentState segmentState, MKBI metaKeyInterop, KBI keyInterop, KB key, long keySize, InstanceOrBytesToInstance toKey, ReadValue readValue, RV usingValue, InstanceOrBytesToInstance toValue, long hash2, boolean create, MutableLockedEntry lock) { + segmentStateNotNullImpliesCopiesNotNull(copies, segmentState); + if (segmentState == null) { + copies = SegmentState.getCopies(copies); + segmentState = SegmentState.get(copies); + } readLock(null); try { return acquireWithoutLock(copies, segmentState, metaKeyInterop, keyInterop, key, keySize, toKey, readValue, usingValue, toValue, hash2, create, lock); } finally { - if (segmentState != null) - segmentState.close(); + segmentState.close(); readUnlock(); } } , RV> RV acquireWithoutLock( - @Nullable ThreadLocalCopies copies, final @Nullable SegmentState segmentState, + @NotNull ThreadLocalCopies copies, @NotNull SegmentState segmentState, MKBI metaKeyInterop, KBI keyInterop, KB key, long keySize, InstanceOrBytesToInstance toKey, ReadValue readValue, RV usingValue, InstanceOrBytesToInstance toValue, long hash2, boolean create, MutableLockedEntry lock) { - segmentStateNotNullImpliesCopiesNotNull(copies, segmentState); - SegmentState localSegmentState = segmentState; - try { - MultiStoreBytes entry = null; - MultiMap hashLookup = this.hashLookup; - hashLookup.startSearch(hash2); - for (long pos; (pos = hashLookup.nextPos()) >= 0L; ) { - long offset = offsetFromPos(pos); - if (entry == null) { - if (localSegmentState == null) { - copies = SegmentState.getCopies(copies); - localSegmentState = SegmentState.get(copies); - } - entry = localSegmentState.tmpBytes; - } - reuse(entry, offset); - if (!keyEquals(keyInterop, metaKeyInterop, key, keySize, entry)) - continue; - // key is found - entry.skip(keySize); - return readValueAndNotifyGet(copies, key, keySize, toKey, - readValue, usingValue, toValue, entry); - } - // key is not found - if (!create) - return readValue.readNull(); - if (entry == null) { - if (localSegmentState == null) { - copies = SegmentState.getCopies(copies); // after this, copies is not null - localSegmentState = SegmentState.get(copies); - } - entry = localSegmentState.tmpBytes; - } - - RV result = createEntryOnAcquire(copies, localSegmentState, - metaKeyInterop, keyInterop, key, keySize, toKey, + MultiStoreBytes entry = segmentState.tmpBytes; + MultiMap hashLookup = this.hashLookup; + SearchState searchState = segmentState.searchState; + hashLookup.startSearch(hash2, searchState); + for (long pos; (pos = hashLookup.nextPos(searchState)) >= 0L; ) { + long offset = offsetFromPos(pos); + reuse(entry, offset); + if (!keyEquals(keyInterop, metaKeyInterop, key, keySize, entry)) + continue; + // key is found + entry.skip(keySize); + return readValueAndNotifyGet(copies, key, keySize, toKey, readValue, usingValue, toValue, entry); - - // notify the context that the entry was created - if (lock instanceof WriteLocked) - ((WriteLocked) lock).created(true); - - return result; - } finally { - if (segmentState == null && localSegmentState != null) - localSegmentState.close(); } + // key is not found + if (!create) + return readValue.readNull(); + + RV result = createEntryOnAcquire(copies, segmentState, + metaKeyInterop, keyInterop, key, keySize, toKey, + readValue, usingValue, toValue, entry); + + // notify the context that the entry was created + if (lock instanceof WriteLocked) + ((WriteLocked) lock).created(true); + + return result; } final , RV> @@ -2056,7 +2042,7 @@ class VanillaChronicleMap, , RV, VB extends RV, VBI, MVBI extends MetaBytesInterop> - RV put3(@Nullable ThreadLocalCopies copies, final @Nullable SegmentState segmentState, + RV put3(@Nullable ThreadLocalCopies copies, @Nullable SegmentState segmentState, MKBI metaKeyInterop, KBI keyInterop, KB key, long keySize, InstanceOrBytesToInstance toKey, GetValueInterops getValueInterops, VB value, @@ -2064,6 +2050,11 @@ class VanillaChronicleMap, long hash2, boolean replaceIfPresent, ReadValue readValue, boolean resultUnused) { shouldNotBeCalledFromReplicatedChronicleMap("Segment.put"); + segmentStateNotNullImpliesCopiesNotNull(copies, segmentState); + if (segmentState == null) { + copies = SegmentState.getCopies(copies); + segmentState = SegmentState.get(copies); + } writeLock(); try { return putWithoutLock(copies, segmentState, @@ -2071,8 +2062,7 @@ class VanillaChronicleMap, getValueInterops, value, toValue, hash2, replaceIfPresent, readValue, resultUnused); } finally { - if (segmentState != null) - segmentState.close(); + segmentState.close(); writeUnlock(); } } @@ -2080,63 +2070,53 @@ class VanillaChronicleMap, , RV, VB extends RV, VBI, MVBI extends MetaBytesInterop> RV putWithoutLock( - @Nullable ThreadLocalCopies copies, final @Nullable SegmentState segmentState, + @NotNull ThreadLocalCopies copies, @NotNull SegmentState segmentState, MKBI metaKeyInterop, KBI keyInterop, KB key, long keySize, InstanceOrBytesToInstance toKey, GetValueInterops getValueInterops, VB value, InstanceOrBytesToInstance toValue, long hash2, boolean replaceIfPresent, ReadValue readValue, boolean resultUnused) { - segmentStateNotNullImpliesCopiesNotNull(copies, segmentState); - SegmentState localSegmentState = segmentState; - if (localSegmentState == null) { - copies = SegmentState.getCopies(copies); // copies is not null now - localSegmentState = SegmentState.get(copies); - } - try { - 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); - if (replaceIfPresent) { - return replaceValueAndNotifyPut(copies, localSegmentState, - key, keySize, toKey, - getValueInterops, value, toValue, - entry, pos, offset, hashLookup, readValue, resultUnused, - false, false); - } else { - long valueSize = readValueSize(entry); - return resultUnused ? null : - readValue.readValue(copies, entry, null, valueSize); - } + SearchState searchState = segmentState.searchState; + hashLookup.startSearch(hash2, searchState); + MultiStoreBytes entry = segmentState.tmpBytes; + for (long pos; (pos = hashLookup.nextPos(searchState)) >= 0L; ) { + long offset = offsetFromPos(pos); + reuse(entry, offset); + if (!keyEquals(keyInterop, metaKeyInterop, key, keySize, entry)) + continue; + // key is found + entry.skip(keySize); + if (replaceIfPresent) { + return replaceValueAndNotifyPut(copies, segmentState, + key, keySize, toKey, + getValueInterops, value, toValue, + entry, pos, offset, hashLookup, readValue, resultUnused, + false, false); + } 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); - - // put callbacks - onPut(this, localSegmentState.pos); - 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(); } + // key is not found + VBI valueInterop = getValueInterops.getValueInterop(copies); + MVBI metaValueInterop = + getValueInterops.getMetaValueInterop(copies, valueInterop, value); + long valueSize = metaValueInterop.size(valueInterop, value); + putEntry(segmentState, metaKeyInterop, keyInterop, key, keySize, + metaValueInterop, valueInterop, value, entry, false); + + // put callbacks + onPut(this, segmentState.pos); + if (bytesEventListener != null) + bytesEventListener.onPut(entry, 0L, metaDataBytes, + segmentState.valueSizePos, true); + if (eventListener != null) + eventListener.onPut(toKey.toInstance(copies, key, keySize), + toValue.toInstance(copies, value, valueSize), null); + + return resultUnused ? null : readValue.readNull(); } final > @@ -2220,7 +2200,7 @@ class VanillaChronicleMap, alignment.alignPositionAddr(entry); metaElemWriter.write(elemWriter, entry, elem); - hashLookup.putAfterFailedSearch(pos); + hashLookup.putAfterFailedSearch(segmentState.searchState, pos); incrementSize(); return valueSize; @@ -2305,6 +2285,11 @@ class VanillaChronicleMap, GetValueInterops getValueInterops, VB expectedValue, InstanceOrBytesToInstance toValue, long hash2, ReadValue readValue, boolean resultUnused) { + segmentStateNotNullImpliesCopiesNotNull(copies, segmentState); + if (segmentState == null) { + copies = SegmentState.getCopies(copies); + segmentState = SegmentState.get(copies); + } writeLock(); try { return removeWithoutLock(copies, segmentState, @@ -2312,8 +2297,7 @@ class VanillaChronicleMap, getValueInterops, expectedValue, toValue, hash2, readValue, resultUnused); } finally { - if (segmentState != null) - segmentState.close(); + segmentState.close(); writeUnlock(); } } @@ -2327,63 +2311,50 @@ class VanillaChronicleMap, , RV, VB extends RV, VBI, MVBI extends MetaBytesInterop> Object removeWithoutLock( - @Nullable ThreadLocalCopies copies, final @Nullable SegmentState segmentState, + @NotNull ThreadLocalCopies copies, @NotNull SegmentState segmentState, MKBI metaKeyInterop, KBI keyInterop, KB key, long keySize, InstanceOrBytesToInstance toKey, GetValueInterops getValueInterops, VB expectedValue, InstanceOrBytesToInstance toValue, long hash2, ReadValue readValue, boolean resultUnused) { - segmentStateNotNullImpliesCopiesNotNull(copies, segmentState); - SegmentState localSegmentState = segmentState; - 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 (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 valueSizePos = entry.position(); - long valueSize = readValueSize(entry); + MultiStoreBytes entry = segmentState.tmpBytes; + SearchState searchState = segmentState.searchState; + MultiMap hashLookup = hashLookup(); + hashLookup.startSearch(hash2, searchState); + for (long pos; (pos = hashLookup.nextPos(searchState)) >= 0L; ) { + long offset = offsetFromPos(pos); + reuse(entry, offset); + if (!keyEquals(keyInterop, metaKeyInterop, key, keySize, entry)) + continue; + // key is found + entry.skip(keySize); + long valueSizePos = entry.position(); + long valueSize = readValueSize(entry); - // 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; - } - return removeEntry(copies, key, keySize, toKey, toValue, - readValue, resultUnused, hashLookup, entry, pos, valueSizePos, - valueSize, false, true, expectedValue != null); + // 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; } - // key is not found - if (expectedValue == null) { - return resultUnused ? null : readValue.readNull(); - } else { - return Boolean.FALSE; - } - } finally { - if (segmentState == null && localSegmentState != null) - localSegmentState.close(); + return removeEntry(copies, searchState, key, keySize, toKey, toValue, + readValue, resultUnused, hashLookup, entry, pos, valueSizePos, + valueSize, false, true, expectedValue != null); + } + // key is not found + if (expectedValue == null) { + return resultUnused ? null : readValue.readNull(); + } else { + return Boolean.FALSE; } } final Object removeEntry( - ThreadLocalCopies copies, + ThreadLocalCopies copies, SearchState searchState, KB key, long keySize, InstanceOrBytesToInstance toKey, InstanceOrBytesToInstance toValue, ReadValue readValue, boolean resultUnused, @@ -2400,7 +2371,7 @@ class VanillaChronicleMap, // update segment state if (removeFromMultiMap) { - hashLookup.removePrevPos(); + hashLookup.removePrevPos(searchState); long entrySizeInBytes = entry.positionAddr() + valueSize - entry.startAddr(); free(pos, inBlocks(entrySizeInBytes)); } else { @@ -2434,18 +2405,14 @@ class VanillaChronicleMap, boolean containsKey(ThreadLocalCopies copies, MKBI metaKeyInterop, KBI keyInterop, KB key, long keySize, long hash2) { readLock(null); - SegmentState segmentState = null; - try { - MultiStoreBytes entry = null; + copies = SegmentState.getCopies(copies); + try (SegmentState segmentState = SegmentState.get(copies)) { + MultiStoreBytes entry = segmentState.tmpBytes; MultiMap hashLookup = hashLookup(); - hashLookup.startSearch(hash2); - for (long pos; (pos = hashLookup.nextPos()) >= 0L; ) { + SearchState searchState = segmentState.searchState; + hashLookup.startSearch(hash2, searchState); + for (long pos; (pos = hashLookup.nextPos(searchState)) >= 0L; ) { long offset = offsetFromPos(pos); - if (entry == null) { - copies = SegmentState.getCopies(copies); - segmentState = SegmentState.get(copies); - entry = segmentState.tmpBytes; - } reuse(entry, offset); if (!keyEquals(keyInterop, metaKeyInterop, key, keySize, entry)) continue; @@ -2455,8 +2422,6 @@ class VanillaChronicleMap, } return false; } finally { - if (segmentState != null) - segmentState.close(); readUnlock(); } } @@ -2484,19 +2449,17 @@ class VanillaChronicleMap, ReadValue readValue, InstanceOrBytesToInstance toValue, long hash2) { segmentStateNotNullImpliesCopiesNotNull(copies, segmentState); + if (segmentState == null) { + copies = SegmentState.getCopies(copies); + segmentState = SegmentState.get(copies); + } writeLock(); try { - hashLookup.startSearch(hash2); - MultiStoreBytes entry = null; - for (long pos; (pos = hashLookup.nextPos()) >= 0L; ) { + SearchState searchState = segmentState.searchState; + hashLookup.startSearch(hash2, searchState); + MultiStoreBytes entry = segmentState.tmpBytes; + for (long pos; (pos = hashLookup.nextPos(searchState)) >= 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; @@ -2510,8 +2473,7 @@ class VanillaChronicleMap, // key is not found return expectedValue == null ? null : Boolean.FALSE; } finally { - if (segmentState != null) - segmentState.close(); + segmentState.close(); writeUnlock(); } } @@ -2608,7 +2570,7 @@ class VanillaChronicleMap, pos = alloc(newSizeInBlocks); // putValue() is called from put() and replace() // after successful search by key - searchedHashLookup.replacePrevPos(pos); + searchedHashLookup.replacePrevPos(segmentState.searchState, pos); offset = offsetFromPos(pos); // Moving metadata, key size and key. // Don't want to fiddle with pseudo-buffers for this, diff --git a/src/test/java/net/openhft/chronicle/map/IntIntMultiMapTest.java b/src/test/java/net/openhft/chronicle/map/IntIntMultiMapTest.java index 907e6238e..44d4418b5 100644 --- a/src/test/java/net/openhft/chronicle/map/IntIntMultiMapTest.java +++ b/src/test/java/net/openhft/chronicle/map/IntIntMultiMapTest.java @@ -82,9 +82,10 @@ public class IntIntMultiMapTest { private void valuesEqualsByKey(long k) { List values = new ArrayList(); - map.startSearch(k); + SearchState searchState = new SearchState(); + map.startSearch(k, searchState); long v; - while ((v = map.nextPos()) >= 0L) + while ((v = map.nextPos(searchState)) >= 0L) values.add(v); Set valueSet = new HashSet(values); Assert.assertEquals(values.size(), valueSet.size()); @@ -146,8 +147,9 @@ public class IntIntMultiMapTest { map.put(15, 1); map.remove(15, 1); - map.startSearch(15); - Assert.assertTrue(map.nextPos() < 0); + SearchState searchState = new SearchState(); + map.startSearch(15, searchState); + Assert.assertTrue(map.nextPos(searchState) < 0); } @Test