diff --git a/src/main/java/net/openhft/chronicle/map/ChannelProvider.java b/src/main/java/net/openhft/chronicle/map/ChannelProvider.java index 383e1eb2c..54a9b74ef 100644 --- a/src/main/java/net/openhft/chronicle/map/ChannelProvider.java +++ b/src/main/java/net/openhft/chronicle/map/ChannelProvider.java @@ -191,7 +191,7 @@ final class ChannelProvider implements Closeable { @Override public boolean nextEntry(@NotNull EntryCallback callback, - final int na) { + final int na) throws InterruptedException { channelDataLock.readLock().lock(); try { for (int i = 0, len = chronicleChannelList.size(); i < len; i++) { @@ -213,9 +213,13 @@ final class ChannelProvider implements Closeable { channelDataLock.readLock().lock(); try { for (int i = 0, len = chronicleChannelList.size(); i < len; i++) { - chronicleChannelList.get(i) - .acquireModificationIterator(remoteIdentifier, notifier) - .dirtyEntries(fromTimeStamp); + try { + chronicleChannelList.get(i) + .acquireModificationIterator(remoteIdentifier, notifier) + .dirtyEntries(fromTimeStamp); + } catch (InterruptedException e) { + throw new RuntimeException(e); + } notifier.onChange(); } } finally { @@ -335,8 +339,12 @@ final class ChannelProvider implements Closeable { // this could be null if one node has a chronicle channel before the other if (chronicleChannels[chronicleChannel] != null) { - chronicleChannels[chronicleChannel].acquireModificationIterator(remoteIdentifier, NOP) - .dirtyEntries(lastModificationTime); + try { + chronicleChannels[chronicleChannel].acquireModificationIterator(remoteIdentifier, NOP) + .dirtyEntries(lastModificationTime); + } catch (InterruptedException e) { + throw new RuntimeException(e); + } } } diff --git a/src/main/java/net/openhft/chronicle/map/Replica.java b/src/main/java/net/openhft/chronicle/map/Replica.java index 1077c44f7..9ae880c6c 100644 --- a/src/main/java/net/openhft/chronicle/map/Replica.java +++ b/src/main/java/net/openhft/chronicle/map/Replica.java @@ -109,7 +109,7 @@ interface Replica extends Closeable { * @return {@code true} if the entry was accepted by the {@code callback.onEntry()} method, * {@code false} if the entry was not accepted or was not available */ - boolean nextEntry(@NotNull final EntryCallback callback, final int chronicleId); + boolean nextEntry(@NotNull final EntryCallback callback, final int chronicleId) throws InterruptedException; /** * Dirties all entries with a modification time equal to {@code fromTimeStamp} or newer. It @@ -119,7 +119,7 @@ interface Replica extends Closeable { * * @param fromTimeStamp the timestamp from which all entries should be dirty */ - void dirtyEntries(long fromTimeStamp); + void dirtyEntries(long fromTimeStamp) throws InterruptedException; } /** diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java index b033d5f8d..6916a913b 100644 --- a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java @@ -1254,7 +1254,7 @@ final class ReplicatedChronicleMap { /** * if passed segmentState is null, null is returned instead of readLock */ - AutoCloseable readLock(@Nullable VanillaChronicleMap.SegmentState segmentState); + AutoCloseable readLock(@Nullable VanillaChronicleMap.SegmentState segmentState) throws InterruptedException; Map.Entry getEntry(@NotNull VanillaChronicleMap.SegmentState segmentState, long pos); diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index b19e3bdad..ffcc12f14 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -541,7 +541,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Clo */ private void doHandShaking(@NotNull final SelectionKey key, @NotNull SocketChannel socketChannel) - throws IOException { + throws IOException, InterruptedException { final Attached attached = (Attached) key.attachment(); final TcpSocketChannelEntryWriter writer = attached.entryWriter; final TcpSocketChannelEntryReader reader = attached.entryReader; @@ -652,7 +652,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Clo * called when the selector receives a OP_WRITE message */ private void onWrite(@NotNull final SelectionKey key, - final long approxTime) throws IOException { + final long approxTime) throws IOException, InterruptedException { final SocketChannel socketChannel = (SocketChannel) key.channel(); final Attached attached = (Attached) key.attachment(); if (attached == null) { @@ -1077,7 +1077,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Clo * @param selectionKey */ void entriesToBuffer(@NotNull final Replica.ModificationIterator modificationIterator, - @NotNull final SelectionKey selectionKey) { + @NotNull final SelectionKey selectionKey) throws InterruptedException { final SocketChannel socketChannel = (SocketChannel) selectionKey.channel(); final Attached attached = (Attached) selectionKey.attachment(); diff --git a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java index c03366aa4..3d7813ef0 100755 --- a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java @@ -235,7 +235,7 @@ class VanillaChronicleMap, this.ms = new CheckedBytesStore(bytesStore); - // this.ms = bytesStore; + // this.ms = bytesStore; onHeaderCreated(); @@ -577,19 +577,31 @@ class VanillaChronicleMap, @SuppressWarnings("unchecked") public final V get(Object key) { checkKey(key); - return lookupUsing((K) key, null, false); + try { + return lookupUsing((K) key, null, false); + } catch (InterruptedException e) { + throw new RuntimeException(e); + } } @Override public final V getUsing(K key, V usingValue) { checkKey(key); - return lookupUsing(key, usingValue, false); + try { + return lookupUsing(key, usingValue, false); + } catch (InterruptedException e) { + throw new RuntimeException(e); + } } @Override public final V acquireUsing(@NotNull K key, V usingValue) { checkKey(key); - return lookupUsing(key, usingValue, true); + try { + return lookupUsing(key, usingValue, true); + } catch (InterruptedException e) { + throw new RuntimeException(e); + } } final void getBytes(Bytes key, TcpReplicator.TcpSocketChannelEntryWriter output) { @@ -597,14 +609,18 @@ class VanillaChronicleMap, SegmentState segmentState = SegmentState.get(copies); ReadValueToOutputBytes readValueToOutputBytes = segmentState.readValueToOutputBytes; readValueToOutputBytes.reuse(valueSizeMarshaller, output); - lookupUsing(copies, segmentState, - DelegatingMetaBytesInterop.>instance(), - BytesBytesInterop.INSTANCE, key, keySizeMarshaller.readSize(key), - keyBytesToInstance, readValueToOutputBytes, null, outputValueBytesToInstance, - false, null); + try { + lookupUsing(copies, segmentState, + DelegatingMetaBytesInterop.>instance(), + BytesBytesInterop.INSTANCE, key, keySizeMarshaller.readSize(key), + keyBytesToInstance, readValueToOutputBytes, null, outputValueBytesToInstance, + false, null); + } catch (InterruptedException e) { + throw new RuntimeException(e); + } } - private V lookupUsing(K key, V usingValue, boolean create) { + private V lookupUsing(K key, V usingValue, boolean create) throws InterruptedException { ThreadLocalCopies copies = keyInteropProvider.getCopies(null); KI keyInterop = keyInteropProvider.get(copies, originalKeyInterop); copies = metaKeyInteropProvider.getCopies(copies); @@ -620,7 +636,7 @@ class VanillaChronicleMap, MKBI metaKeyInterop, KBI keyInterop, KB key, long keySize, InstanceOrBytesToInstance toKey, ReadValue readValue, RV usingValue, InstanceOrBytesToInstance toValue, - boolean create, final MutableLockedEntry lock) { + boolean create, final MutableLockedEntry lock) throws InterruptedException { long hash = metaKeyInterop.hash(keyInterop, key); int segmentNum = getSegment(hash); long segmentHash = segmentHash(hash); @@ -635,6 +651,8 @@ class VanillaChronicleMap, try (ReadContext entry = lookupUsing(key, null, false, false, LockType.READ_LOCK)) { return entry.present() ? function.apply(entry.value()) : null; + } catch (InterruptedException e) { + throw new RuntimeException(e); } } @@ -662,7 +680,10 @@ class VanillaChronicleMap, usingSB.append((CharSequence) result); } return result; + } catch (InterruptedException e) { + throw new RuntimeException(e); } + } V putMapped0(K key, @NotNull UnaryOperator unaryOperator) { @@ -675,6 +696,8 @@ class VanillaChronicleMap, V result = unaryOperator.update(entry.value()); ((MutableLockedEntry) entry).value(result); return result; + } catch (InterruptedException e) { + throw new RuntimeException(e); } } @@ -750,17 +773,25 @@ class VanillaChronicleMap, @NotNull @Override public final WriteContext acquireUsingLocked(@NotNull K key, @NotNull V usingValue) { - return lookupUsing(key, usingValue, true, true, LockType.WRITE_LOCK); + try { + return lookupUsing(key, usingValue, true, true, LockType.WRITE_LOCK); + } catch (InterruptedException e) { + throw new RuntimeException(e); + } } @NotNull @Override public final ReadContext getUsingLocked(@NotNull K key, @NotNull V usingValue) { - return lookupUsing(key, usingValue, true, false, LockType.READ_LOCK); + try { + return lookupUsing(key, usingValue, true, false, LockType.READ_LOCK); + } catch (InterruptedException e) { + throw new RuntimeException(e); + } } private T lookupUsing( - K key, V usingValue, boolean mustReuseValue, final boolean create, final LockType lockType) { + K key, V usingValue, boolean mustReuseValue, final boolean create, final LockType lockType) throws InterruptedException { checkKey(key); ThreadLocalCopies copies = SegmentState.getCopies(null); @@ -857,11 +888,15 @@ class VanillaChronicleMap, copies = metaKeyInteropProvider.getCopies(copies); MKI metaKeyInterop = metaKeyInteropProvider.get(copies, originalMetaKeyInterop, keyInterop, key); - return containsKey(copies, metaKeyInterop, keyInterop, key, - metaKeyInterop.size(keyInterop, key)); + try { + return containsKey(copies, metaKeyInterop, keyInterop, key, + metaKeyInterop.size(keyInterop, key)); + } catch (InterruptedException e) { + throw new RuntimeException(e); + } } - final boolean containsBytesKey(Bytes key) { + final boolean containsBytesKey(Bytes key) throws InterruptedException { return containsKey(null, DelegatingMetaBytesInterop.>instance(), BytesBytesInterop.INSTANCE, @@ -870,7 +905,7 @@ class VanillaChronicleMap, private > boolean containsKey(ThreadLocalCopies copies, - MKBI metaKeyInterop, KBI keyInterop, KB key, long keySize) { + MKBI metaKeyInterop, KBI keyInterop, KB key, long keySize) throws InterruptedException { long hash = metaKeyInterop.hash(keyInterop, key); int segmentNum = getSegment(hash); long segmentHash = segmentHash(hash); @@ -1221,7 +1256,7 @@ class VanillaChronicleMap, /** * For testing */ - final void checkConsistency() { + final void checkConsistency() throws InterruptedException { for (Segment segment : segments) { segment.checkConsistency(); } @@ -1534,15 +1569,13 @@ class VanillaChronicleMap, @Override public final ReadLocked readLock( - @Nullable SegmentState segmentState) { + @Nullable SegmentState segmentState) throws InterruptedException { while (true) { // final boolean success = segmentHeader.tryRWReadLock(LOCK_OFFSET, lockTimeOutNS); final boolean success; - try { - success = segmentHeader.tryRWWriteLock(LOCK_OFFSET, lockTimeOutNS); - } catch (InterruptedException e) { - throw new IllegalStateException(e); - } + + success = segmentHeader.tryRWWriteLock(LOCK_OFFSET, lockTimeOutNS); + /* boolean success = false; try { @@ -1687,7 +1720,7 @@ class VanillaChronicleMap, MKBI metaKeyInterop, KBI keyInterop, KB key, long keySize, InstanceOrBytesToInstance toKey, ReadValue readValue, RV usingValue, InstanceOrBytesToInstance toValue, - long hash2, boolean create, MutableLockedEntry lock) { + long hash2, boolean create, MutableLockedEntry lock) throws InterruptedException { segmentStateNotNullImpliesCopiesNotNull(copies, segmentState); if (segmentState == null) { copies = SegmentState.getCopies(copies); @@ -2194,7 +2227,7 @@ class VanillaChronicleMap, private > boolean containsKey(ThreadLocalCopies copies, - MKBI metaKeyInterop, KBI keyInterop, KB key, long keySize, long hash2) { + MKBI metaKeyInterop, KBI keyInterop, KB key, long keySize, long hash2) throws InterruptedException { readLock(null); copies = SegmentState.getCopies(copies); try (SegmentState segmentState = SegmentState.get(copies)) { @@ -2437,7 +2470,7 @@ class VanillaChronicleMap, /** * Check there is no garbage in freeList. */ - private void checkConsistency() { + private void checkConsistency() throws InterruptedException { readLock(null); try (SegmentState segmentState = SegmentState.get(SegmentState.getCopies(null))) { MultiStoreBytes entry = segmentState.tmpBytes; @@ -2530,7 +2563,11 @@ class VanillaChronicleMap, ThreadLocalCopies copies = SegmentState.getCopies(null); SegmentState segmentState = SegmentState.get(copies); try { - segment.readLock(null); + try { + segment.readLock(null); + } catch (InterruptedException e) { + throw new RuntimeException(e); + } if (segment.hashLookup().getPositions().isClear(pos)) { // the pos was removed after the previous advance advance(segIndex, pos); diff --git a/src/test/java/net/openhft/chronicle/map/ChronicleMapTest.java b/src/test/java/net/openhft/chronicle/map/ChronicleMapTest.java index 6fa4ca887..f17bfcf43 100755 --- a/src/test/java/net/openhft/chronicle/map/ChronicleMapTest.java +++ b/src/test/java/net/openhft/chronicle/map/ChronicleMapTest.java @@ -1679,7 +1679,7 @@ public class ChronicleMapTest { @Test - public void testEntriesSpanningSeveralChunksPutRemoveReplace() throws IOException { + public void testEntriesSpanningSeveralChunksPutRemoveReplace() throws IOException, InterruptedException { ChronicleMapBuilder builder = ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) .entries(10) .minSegments(1)