From ceab61fd85e2d64264bae68910259abd41679d89 Mon Sep 17 00:00:00 2001 From: Peter Lawrey Date: Tue, 12 Apr 2016 11:39:23 +0100 Subject: [PATCH 1/3] Rename Not Ready to Not Complete. Support reading Not COmplete messages unless being read from shared memory. --- .../chronicle/hash/impl/SizePrefixedBlob.java | 2 +- .../openhft/chronicle/map/ChronicleMapBuilder.java | 12 ++++++------ 2 files changed, 7 insertions(+), 7 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/hash/impl/SizePrefixedBlob.java b/src/main/java/net/openhft/chronicle/hash/impl/SizePrefixedBlob.java index 7547d6ef5..d4d47348d 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/SizePrefixedBlob.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/SizePrefixedBlob.java @@ -23,7 +23,7 @@ public final class SizePrefixedBlob { public static final int SELF_BOOTSTRAPPING_HEADER_OFFSET = 12; public static final int READY = 0; - public static final int NOT_READY = 0x80000000; + public static final int NOT_COMPLETE = 0x80000000; public static final int DATA = 0; public static final int META_DATA = 0x40000000; diff --git a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java index f44b62d3e..c29f3297d 100755 --- a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java @@ -1544,7 +1544,7 @@ public final class ChronicleMapBuilder implements int sizeWord = headerBuffer.getInt(SIZE_WORD_OFFSET); if (!SizePrefixedBlob.isReady(sizeWord)) { if (recover) { - LOG.error("size-prefixed blob readiness bit is set to NOT_READY"); + LOG.error("size-prefixed blob readiness bit is set to NOT_COMPLETE"); // the bit will be overwritten to READY in the end of recovery procedure, so nothing // to fix right here } else { @@ -1609,7 +1609,7 @@ public final class ChronicleMapBuilder implements map.createMappedStoreAndSegments(); } else { if (!overrideBuilderConfig) - writeNotReady(fileChannel, headerBuffer, headerSize); + writeNotComplete(fileChannel, headerBuffer, headerSize); // if overrideBuilderConfig = true, readiness bit is already set // in writeHeader() call map.recover(); @@ -1625,10 +1625,10 @@ public final class ChronicleMapBuilder implements } } - private static void writeNotReady( + private static void writeNotComplete( FileChannel fileChannel, ByteBuffer headerBuffer, int headerSize) throws IOException { //noinspection PointlessBitwiseExpression - headerBuffer.putInt(SIZE_WORD_OFFSET, NOT_READY | DATA | headerSize); + headerBuffer.putInt(SIZE_WORD_OFFSET, NOT_COMPLETE | DATA | headerSize); headerBuffer.clear().position(SIZE_WORD_OFFSET).limit(SIZE_WORD_OFFSET + 4); writeFully(fileChannel, SIZE_WORD_OFFSET, headerBuffer); } @@ -1656,10 +1656,10 @@ public final class ChronicleMapBuilder implements long checksum = headerChecksum(headerBuffer, headerSize); headerBuffer.putLong(HEADER_OFFSET, checksum); - // Set readiness bit to NOT_READY, because the Chronicle Map instance is not actually + // Set readiness bit to NOT_COMPLETE, because the Chronicle Map instance is not actually // ready yet //noinspection PointlessBitwiseExpression - headerBuffer.putInt(SIZE_WORD_OFFSET, NOT_READY | DATA | headerSize); + headerBuffer.putInt(SIZE_WORD_OFFSET, NOT_COMPLETE | DATA | headerSize); // Write the size-prefixed blob to the file headerBuffer.position(0).limit(headerLimit); From ffff33bed1ada0450f3ed3ce492d5de8bab16cb4 Mon Sep 17 00:00:00 2001 From: Peter Lawrey Date: Tue, 12 Apr 2016 12:08:14 +0100 Subject: [PATCH 2/3] Reduce Thread.sleep to Jvm.pause, delays >= 1s to 500 ms, and remove "throws Exception" from tests --- .../java/net/openhft/chronicle/map/ChronicleMapBuilder.java | 2 +- .../java/net/openhft/chronicle/map/CHMLatencyTestMain.java | 2 +- src/test/java/net/openhft/chronicle/map/CHMTest5.java | 2 +- .../java/net/openhft/chronicle/map/CHMUseCasesTest.java | 2 +- .../openhft/chronicle/map/ChronicleMapSanityCheckTest.java | 2 +- .../net/openhft/chronicle/map/Issue62ChronicleClient.java | 2 +- .../net/openhft/chronicle/map/Issue62ChronicleServer.java | 2 +- src/test/java/net/openhft/chronicle/map/Issue63Test.java | 6 +++--- .../java/net/openhft/chronicle/map/LargeEntriesTest.java | 2 +- .../openhft/chronicle/map/ProcessInstanceLimiterMain.java | 4 ++-- .../net/openhft/chronicle/map/ReplicationCheckingMap.java | 2 +- .../chronicle/map/example/PointListSerializationTest.java | 2 +- .../net/openhft/chronicle/map/jsr166/JSR166TestCase.java | 2 +- .../java/net/openhft/chronicle/map/jsr166/map/MapCheck.java | 2 +- .../chronicle/map/utility/ProcessInstanceLimiter.java | 6 +++--- src/test/java/net/openhft/chronicle/set/Builder.java | 4 ++-- 16 files changed, 22 insertions(+), 22 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java index c29f3297d..6354d9118 100755 --- a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java @@ -1501,7 +1501,7 @@ public final class ChronicleMapBuilder implements // truncated between length() and read() calls, then continue to wait } try { - Thread.sleep(100); + Jvm.pause(100); } catch (InterruptedException e) { if (recover) { break; diff --git a/src/test/java/net/openhft/chronicle/map/CHMLatencyTestMain.java b/src/test/java/net/openhft/chronicle/map/CHMLatencyTestMain.java index f169d9bac..a52653577 100755 --- a/src/test/java/net/openhft/chronicle/map/CHMLatencyTestMain.java +++ b/src/test/java/net/openhft/chronicle/map/CHMLatencyTestMain.java @@ -144,7 +144,7 @@ public class CHMLatencyTestMain { public void run() { while (running) { try { - Thread.sleep(1); + Jvm.pause(1); } catch (InterruptedException e) { e.printStackTrace(); break; diff --git a/src/test/java/net/openhft/chronicle/map/CHMTest5.java b/src/test/java/net/openhft/chronicle/map/CHMTest5.java index 58f377c1f..d17a24237 100644 --- a/src/test/java/net/openhft/chronicle/map/CHMTest5.java +++ b/src/test/java/net/openhft/chronicle/map/CHMTest5.java @@ -233,7 +233,7 @@ public class CHMTest5 { long elapsedTime; while ((elapsedTime = System.currentTimeMillis() - start) < pause) { try { - Thread.sleep(pause - elapsedTime); + Jvm.pause(pause - elapsedTime); } catch (InterruptedException e) { } } diff --git a/src/test/java/net/openhft/chronicle/map/CHMUseCasesTest.java b/src/test/java/net/openhft/chronicle/map/CHMUseCasesTest.java index cb1b9dd35..8279de769 100644 --- a/src/test/java/net/openhft/chronicle/map/CHMUseCasesTest.java +++ b/src/test/java/net/openhft/chronicle/map/CHMUseCasesTest.java @@ -115,7 +115,7 @@ public class CHMUseCasesTest { if (map1.equals(map2)) break; try { - Thread.sleep(1); + Jvm.pause(1); } catch (InterruptedException e) { throw new RuntimeException(e); } diff --git a/src/test/java/net/openhft/chronicle/map/ChronicleMapSanityCheckTest.java b/src/test/java/net/openhft/chronicle/map/ChronicleMapSanityCheckTest.java index 84a7500af..6339b343e 100644 --- a/src/test/java/net/openhft/chronicle/map/ChronicleMapSanityCheckTest.java +++ b/src/test/java/net/openhft/chronicle/map/ChronicleMapSanityCheckTest.java @@ -107,7 +107,7 @@ public class ChronicleMapSanityCheckTest { }, 0, consumerPeriod, consumerTimeUnit); - Thread.sleep(totalTestTimeMS); + Jvm.pause(totalTestTimeMS); consumerExecutor.shutdown(); try { diff --git a/src/test/java/net/openhft/chronicle/map/Issue62ChronicleClient.java b/src/test/java/net/openhft/chronicle/map/Issue62ChronicleClient.java index 300c820b4..a5abd1be6 100644 --- a/src/test/java/net/openhft/chronicle/map/Issue62ChronicleClient.java +++ b/src/test/java/net/openhft/chronicle/map/Issue62ChronicleClient.java @@ -45,7 +45,7 @@ public class Issue62ChronicleClient { cityPostalCodesMapBuilder.createPersistedTo(MAP_FILE_B)) { LOGGER.info("Starting"); - Thread.sleep(3000); + Jvm.pause(3000); for (Entry entry : map.entrySet()) { LOGGER.info("{} : {}", entry.getKey(), entry.getValue()); diff --git a/src/test/java/net/openhft/chronicle/map/Issue62ChronicleServer.java b/src/test/java/net/openhft/chronicle/map/Issue62ChronicleServer.java index 9fada2820..bf8818bd9 100644 --- a/src/test/java/net/openhft/chronicle/map/Issue62ChronicleServer.java +++ b/src/test/java/net/openhft/chronicle/map/Issue62ChronicleServer.java @@ -63,7 +63,7 @@ public class Issue62ChronicleServer { } LOGGER.info("Map created"); - Thread.sleep(15000); + Jvm.pause(15000); } diff --git a/src/test/java/net/openhft/chronicle/map/Issue63Test.java b/src/test/java/net/openhft/chronicle/map/Issue63Test.java index 3c890b2b6..4fbad3478 100644 --- a/src/test/java/net/openhft/chronicle/map/Issue63Test.java +++ b/src/test/java/net/openhft/chronicle/map/Issue63Test.java @@ -39,7 +39,7 @@ import static org.junit.Assert.assertEquals; public class Issue63Test { @Test - public void issue63Test() throws Exception { + public void issue63test() { Path path = Paths.get(System.getProperty("java.io.tmpdir") + "/test-vectors1.dat"); if (Files.exists(path)) Files.delete(path); File mapFile = path.toFile(); @@ -148,7 +148,7 @@ public class Issue63Test { new Issue63Test().testChronicleMap(); } - void testChronicleMap() throws Exception { + void testChronicleMap() { int num = 1_000_000; testChronicleMap(System.getProperty("java.io.tmpdir"), num, num); ThreadLocalRandom random = ThreadLocalRandom.current(); @@ -307,7 +307,7 @@ public class Issue63Test { } @Test - public void testKnownItems() throws Exception { + public void testKnownItems() { ArrayList averageKnownItems = new ArrayList<>(); for (int i = 0; i < 100; i++) { diff --git a/src/test/java/net/openhft/chronicle/map/LargeEntriesTest.java b/src/test/java/net/openhft/chronicle/map/LargeEntriesTest.java index 3d7bfeaa9..9afa0c4b4 100755 --- a/src/test/java/net/openhft/chronicle/map/LargeEntriesTest.java +++ b/src/test/java/net/openhft/chronicle/map/LargeEntriesTest.java @@ -157,7 +157,7 @@ public class LargeEntriesTest { sb.append(thisThread).append("\n"); try { while (!Thread.currentThread().isInterrupted()) { - Thread.sleep(50); + Jvm.pause(50); StackTraceElement[] stackTrace = thisThread.getStackTrace(); for (int i = 0; i < 4 && i < stackTrace.length; i++) sb.append("\tat ").append(stackTrace[i]).append("\n"); diff --git a/src/test/java/net/openhft/chronicle/map/ProcessInstanceLimiterMain.java b/src/test/java/net/openhft/chronicle/map/ProcessInstanceLimiterMain.java index 77e80b9b1..6977e8f8b 100644 --- a/src/test/java/net/openhft/chronicle/map/ProcessInstanceLimiterMain.java +++ b/src/test/java/net/openhft/chronicle/map/ProcessInstanceLimiterMain.java @@ -60,7 +60,7 @@ public class ProcessInstanceLimiterMain implements Runnable { ProcessInstanceLimiterMain limiter = new ProcessInstanceLimiterMain("test", callback); limiter.setMaxNumberOfProcessesOfType("x", 2); limiter.startingProcessOfType("x"); - Thread.sleep(60L * 1000L); + Jvm.pause(60L * 1000L); } public static void pause(long pause) { @@ -68,7 +68,7 @@ public class ProcessInstanceLimiterMain implements Runnable { long elapsedTime; while ((elapsedTime = System.currentTimeMillis() - start) < pause) { try { - Thread.sleep(pause - elapsedTime); + Jvm.pause(pause - elapsedTime); } catch (InterruptedException e) { } } diff --git a/src/test/java/net/openhft/chronicle/map/ReplicationCheckingMap.java b/src/test/java/net/openhft/chronicle/map/ReplicationCheckingMap.java index dadfe9211..3642e8f2d 100644 --- a/src/test/java/net/openhft/chronicle/map/ReplicationCheckingMap.java +++ b/src/test/java/net/openhft/chronicle/map/ReplicationCheckingMap.java @@ -92,7 +92,7 @@ public class ReplicationCheckingMap implements ChronicleMap { if (i > 30) { try { - Thread.sleep(i); + Jvm.pause(i); } catch (InterruptedException e) { e.printStackTrace(); } diff --git a/src/test/java/net/openhft/chronicle/map/example/PointListSerializationTest.java b/src/test/java/net/openhft/chronicle/map/example/PointListSerializationTest.java index 2009b99e7..90c8d436d 100644 --- a/src/test/java/net/openhft/chronicle/map/example/PointListSerializationTest.java +++ b/src/test/java/net/openhft/chronicle/map/example/PointListSerializationTest.java @@ -101,7 +101,7 @@ public class PointListSerializationTest { } @Test - public void testComplexSerialization() throws Exception { + public void testComplexSerialization() { try (ChronicleMap map = ChronicleMapBuilder .of(String.class, A.class) .valueMarshaller(AMarshaller.INSTANCE) diff --git a/src/test/java/net/openhft/chronicle/map/jsr166/JSR166TestCase.java b/src/test/java/net/openhft/chronicle/map/jsr166/JSR166TestCase.java index 7538a8f11..38578a4b7 100644 --- a/src/test/java/net/openhft/chronicle/map/jsr166/JSR166TestCase.java +++ b/src/test/java/net/openhft/chronicle/map/jsr166/JSR166TestCase.java @@ -153,7 +153,7 @@ public class JSR166TestCase { long ns = millis * 1000 * 1000; for (; ; ) { if (millis > 0L) - Thread.sleep(millis); + Jvm.pause(millis); else // too short to sleep Thread.yield(); long d = ns - (System.nanoTime() - startTime); diff --git a/src/test/java/net/openhft/chronicle/map/jsr166/map/MapCheck.java b/src/test/java/net/openhft/chronicle/map/jsr166/map/MapCheck.java index f2c1855be..5c3068bbc 100644 --- a/src/test/java/net/openhft/chronicle/map/jsr166/map/MapCheck.java +++ b/src/test/java/net/openhft/chronicle/map/jsr166/map/MapCheck.java @@ -97,7 +97,7 @@ public class MapCheck { mainTest(key, absent); if ((rep & 3) == 3 && rep < numTests - 1) { shuffle(key); - // Thread.sleep(10); + // Jvm.pause(10); } } diff --git a/src/test/java/net/openhft/chronicle/map/utility/ProcessInstanceLimiter.java b/src/test/java/net/openhft/chronicle/map/utility/ProcessInstanceLimiter.java index 9de4a5bf8..7f6f3fe88 100644 --- a/src/test/java/net/openhft/chronicle/map/utility/ProcessInstanceLimiter.java +++ b/src/test/java/net/openhft/chronicle/map/utility/ProcessInstanceLimiter.java @@ -151,7 +151,7 @@ public class ProcessInstanceLimiter implements Runnable { public static void main(String[] args) throws IOException, InterruptedException { ProcessInstanceLimiter.limitTo(2); - Thread.sleep(60L * 1000L); + Jvm.pause(60L * 1000L); } /** @@ -224,7 +224,7 @@ public class ProcessInstanceLimiter implements Runnable { long elapsedTime; while ((elapsedTime = System.currentTimeMillis() - start) < pause) { try { - Thread.sleep(pause - elapsedTime); + Jvm.pause(pause - elapsedTime); } catch (InterruptedException e) { } } @@ -336,7 +336,7 @@ public class ProcessInstanceLimiter implements Runnable { this.callback.tooManyProcessesOfType(processType); return; } - //try {Thread.sleep(60L*1000L);} catch (InterruptedException e) {} + //try {Jvm.pause(60L*1000L);} catch (InterruptedException e) {} //we've got the lock, now copy the array try { for (int i = 0; i < times1.length; i++) { diff --git a/src/test/java/net/openhft/chronicle/set/Builder.java b/src/test/java/net/openhft/chronicle/set/Builder.java index dd52aec8d..96dbf546e 100644 --- a/src/test/java/net/openhft/chronicle/set/Builder.java +++ b/src/test/java/net/openhft/chronicle/set/Builder.java @@ -59,14 +59,14 @@ public class Builder { // not map1.equals(map2), the reason is described above if (map1.equals(map2)) { numberOfTimesTheSame++; - Thread.sleep(1); + Jvm.pause(1); if (numberOfTimesTheSame == 10) { System.out.println("same"); break; } } - Thread.sleep(1); + Jvm.pause(1); if (System.currentTimeMillis() - startTime > timeOutMs) break; } From b90ebc3b3946f79500c0ed62225d8d91cb37da97 Mon Sep 17 00:00:00 2001 From: Peter Lawrey Date: Fri, 15 Apr 2016 13:09:59 +0100 Subject: [PATCH 3/3] Fix up some tests broken from a tidy up of exceptions. --- pom.xml | 5 +- .../chronicle/map/ChronicleMapBuilder.java | 405 ++++++++-------- .../chronicle/map/CHMLatencyTestMain.java | 14 +- .../net/openhft/chronicle/map/CHMTest5.java | 11 +- .../chronicle/map/CHMUseCasesTest.java | 440 +++++++++--------- .../map/ChronicleMapSanityCheckTest.java | 9 +- .../chronicle/map/Issue62ChronicleClient.java | 1 + .../chronicle/map/Issue62ChronicleServer.java | 9 +- .../openhft/chronicle/map/Issue63Test.java | 102 ++-- .../map/ProcessInstanceLimiterMain.java | 12 +- .../chronicle/map/ReplicationCheckingMap.java | 8 +- .../map/utility/ProcessInstanceLimiter.java | 46 +- .../net/openhft/chronicle/set/Builder.java | 4 +- 13 files changed, 516 insertions(+), 550 deletions(-) diff --git a/pom.xml b/pom.xml index ab960c92f..2396b15c5 100755 --- a/pom.xml +++ b/pom.xml @@ -15,7 +15,8 @@ ~ along with this program. If not, see . --> - + net.openhft @@ -46,7 +47,7 @@ net.openhft chronicle-bom pom - 1.11.16 + 1.12.10 import diff --git a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java index 6354d9118..305832252 100755 --- a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java @@ -135,6 +135,62 @@ public final class ChronicleMapBuilder implements LoggerFactory.getLogger(ChronicleMapBuilder.class.getName()); private static final double UNDEFINED_DOUBLE_CONFIG = Double.NaN; + private static final ConcurrentHashMap fileLockingControl = + new ConcurrentHashMap<>(128); + private static int MAX_BOOTSTRAPPING_HEADER_SIZE = (int) MemoryUnit.KILOBYTES.toBytes(16); + SerializationBuilder keyBuilder; + SerializationBuilder valueBuilder; + K averageKey; + V averageValue; + /** + * Default timeout is 1 minute. Even loopback tests converge often in the course of seconds, + * let alone WAN replication over many nodes might take tens of seconds. + *

+ * TODO review + */ + long cleanupTimeout = 1; + TimeUnit cleanupTimeoutUnit = TimeUnit.MINUTES; + boolean cleanupRemovedEntries = true; + + ////////////////////////////// + // Configuration fields + DefaultValueProvider defaultValueProvider = DefaultSpi.defaultValueProvider(); + byte replicationIdentifier = -1; + MapMethods methods = DefaultSpi.mapMethods(); + MapEntryOperations entryOperations = mapEntryOperations(); + MapRemoteOperations remoteOperations = mapRemoteOperations(); + // not final because of cloning + private ChronicleMapBuilderPrivateAPI privateAPI = + new ChronicleMapBuilderPrivateAPI<>(this); + // used when configuring the number of segments. + private int minSegments = -1; + private int actualSegments = -1; + // used when reading the number of entries per + private long entriesPerSegment = -1L; + private long actualChunksPerSegmentTier = -1L; + private double averageKeySize = UNDEFINED_DOUBLE_CONFIG; + private K sampleKey; + private double averageValueSize = UNDEFINED_DOUBLE_CONFIG; + private V sampleValue; + private int actualChunkSize = 0; + private int worstAlignment = -1; + private int maxChunksPerEntry = -1; + private int alignment = UNDEFINED_ALIGNMENT_CONFIG; + private long entries = -1L; + private double maxBloatFactor = 1.0; + private boolean allowSegmentTiering = true; + private double nonTieredSegmentsPercentile = 0.99999; + private boolean aligned64BitMemoryOperationsAtomic = OS.is64Bit(); + private ChecksumEntries checksumEntries = ChecksumEntries.IF_PERSISTED; + private boolean putReturnsNull = false; + private boolean removeReturnsNull = false; + private boolean replicated; + private boolean persisted; + + ChronicleMapBuilder(Class keyClass, Class valueClass) { + keyBuilder = new SerializationBuilder<>(keyClass); + valueBuilder = new SerializationBuilder<>(valueClass); + } private static boolean isDefined(double config) { return !isNaN(config); @@ -155,14 +211,8 @@ public final class ChronicleMapBuilder implements return (long) v; } - private static int MAX_BOOTSTRAPPING_HEADER_SIZE = (int) MemoryUnit.KILOBYTES.toBytes(16); - - private static final ConcurrentHashMap fileLockingControl = - new ConcurrentHashMap<>(128); - - interface FileIOAction { - void fileIOAction() throws IOException; - } + ////////////////////////////// + // Instance fields /** * When Chronicle Maps are created using {@link #createPersistedTo(File)} or @@ -189,73 +239,6 @@ public final class ChronicleMapBuilder implements }); } - // not final because of cloning - private ChronicleMapBuilderPrivateAPI privateAPI = - new ChronicleMapBuilderPrivateAPI<>(this); - - ////////////////////////////// - // Configuration fields - - SerializationBuilder keyBuilder; - SerializationBuilder valueBuilder; - - // used when configuring the number of segments. - private int minSegments = -1; - private int actualSegments = -1; - // used when reading the number of entries per - private long entriesPerSegment = -1L; - private long actualChunksPerSegmentTier = -1L; - private double averageKeySize = UNDEFINED_DOUBLE_CONFIG; - K averageKey; - private K sampleKey; - private double averageValueSize = UNDEFINED_DOUBLE_CONFIG; - V averageValue; - private V sampleValue; - private int actualChunkSize = 0; - private int worstAlignment = -1; - private int maxChunksPerEntry = -1; - private int alignment = UNDEFINED_ALIGNMENT_CONFIG; - private long entries = -1L; - private double maxBloatFactor = 1.0; - private boolean allowSegmentTiering = true; - private double nonTieredSegmentsPercentile = 0.99999; - private boolean aligned64BitMemoryOperationsAtomic = OS.is64Bit(); - - enum ChecksumEntries {YES, NO, IF_PERSISTED} - private ChecksumEntries checksumEntries = ChecksumEntries.IF_PERSISTED; - - private boolean putReturnsNull = false; - private boolean removeReturnsNull = false; - - /** - * Default timeout is 1 minute. Even loopback tests converge often in the course of seconds, - * let alone WAN replication over many nodes might take tens of seconds. - * - * TODO review - */ - long cleanupTimeout = 1; - TimeUnit cleanupTimeoutUnit = TimeUnit.MINUTES; - boolean cleanupRemovedEntries = true; - - DefaultValueProvider defaultValueProvider = DefaultSpi.defaultValueProvider(); - - byte replicationIdentifier = -1; - - MapMethods methods = DefaultSpi.mapMethods(); - MapEntryOperations entryOperations = mapEntryOperations(); - MapRemoteOperations remoteOperations = mapRemoteOperations(); - - ////////////////////////////// - // Instance fields - - private boolean replicated; - private boolean persisted; - - ChronicleMapBuilder(Class keyClass, Class valueClass) { - keyBuilder = new SerializationBuilder<>(keyClass); - valueBuilder = new SerializationBuilder<>(valueClass); - } - /** * Returns a new {@code ChronicleMapBuilder} instance which is able to {@linkplain #create() * create} maps with the specified key and value classes. @@ -292,6 +275,125 @@ public final class ChronicleMapBuilder implements return obj != null ? obj + "" : "not configured"; } + private static void checkSizeIsStaticallyKnown(SerializationBuilder builder, String role) { + if (builder.sizeIsStaticallyKnown) { + throw new IllegalStateException("Size of " + builder.tClass + + " instances is constant and statically known, shouldn't be specified via " + + "average" + role + "Size() or average" + role + "() methods"); + } + } + + private static void checkAverageSize(double averageSize, String role) { + if (averageSize <= 0 || isNaN(averageSize) || + Double.isInfinite(averageSize)) { + throw new IllegalArgumentException("Average " + role + " size must be a positive, " + + "finite number"); + } + } + + private static double averageSizeStoringLength( + SerializationBuilder builder, double averageSize) { + SizeMarshaller sizeMarshaller = builder.sizeMarshaller(); + if (averageSize == round(averageSize)) + return sizeMarshaller.storingLength(round(averageSize)); + long lower = roundDown(averageSize); + long upper = lower + 1; + int lowerStoringLength = sizeMarshaller.storingLength(lower); + int upperStoringLength = sizeMarshaller.storingLength(upper); + if (lowerStoringLength == upperStoringLength) + return lowerStoringLength; + return lower * (upper - averageSize) + upper * (averageSize - lower); + } + + static int greatestCommonDivisor(int a, int b) { + if (b == 0) return a; + return greatestCommonDivisor(b, a % b); + } + + private static int maxDefaultChunksPerAverageEntry(boolean replicated) { + // When replicated, having 8 chunks (=> 8 bits in bitsets) per entry seems more wasteful + // because when replicated we have bit sets per each remote node, not only allocation + // bit set as when non-replicated + return replicated ? 4 : 8; + } + + private static int estimateSegmentsForEntries(long size) { + if (size > 200 << 20) + return 256; + if (size >= 1 << 20) + return 128; + if (size >= 128 << 10) + return 64; + if (size >= 16 << 10) + return 32; + if (size >= 4 << 10) + return 16; + if (size >= 1 << 10) + return 8; + return 1; + } + + private static long headerChecksum(ByteBuffer headerBuffer, int headerSize) { + return LongHashFunction.xx_r39().hashBytes(headerBuffer, SIZE_WORD_OFFSET, headerSize + 4); + } + + private static void writeNotComplete( + FileChannel fileChannel, ByteBuffer headerBuffer, int headerSize) throws IOException { + //noinspection PointlessBitwiseExpression + headerBuffer.putInt(SIZE_WORD_OFFSET, NOT_COMPLETE | DATA | headerSize); + headerBuffer.clear().position(SIZE_WORD_OFFSET).limit(SIZE_WORD_OFFSET + 4); + writeFully(fileChannel, SIZE_WORD_OFFSET, headerBuffer); + } + + /** + * @return ByteBuffer, with self bootstrapping header in [position, limit) range + */ + private static ByteBuffer writeHeader( + FileChannel fileChannel, VanillaChronicleMap map) throws IOException { + ByteBuffer headerBuffer = ByteBuffer.allocate( + SELF_BOOTSTRAPPING_HEADER_OFFSET + MAX_BOOTSTRAPPING_HEADER_SIZE); + headerBuffer.order(LITTLE_ENDIAN); + + Bytes headerBytes = Bytes.wrapForWrite(headerBuffer); + headerBytes.writePosition(SELF_BOOTSTRAPPING_HEADER_OFFSET); + Wire wire = new TextWire(headerBytes); + wire.getValueOut().typedMarshallable(map); + + int headerLimit = (int) headerBytes.writePosition(); + int headerSize = headerLimit - SELF_BOOTSTRAPPING_HEADER_OFFSET; + // First set readiness bit to READY, to compute checksum correctly + //noinspection PointlessBitwiseExpression + headerBuffer.putInt(SIZE_WORD_OFFSET, READY | DATA | headerSize); + + long checksum = headerChecksum(headerBuffer, headerSize); + headerBuffer.putLong(HEADER_OFFSET, checksum); + + // Set readiness bit to NOT_COMPLETE, because the Chronicle Map instance is not actually + // ready yet + //noinspection PointlessBitwiseExpression + headerBuffer.putInt(SIZE_WORD_OFFSET, NOT_COMPLETE | DATA | headerSize); + + // Write the size-prefixed blob to the file + headerBuffer.position(0).limit(headerLimit); + writeFully(fileChannel, 0, headerBuffer); + + headerBuffer.position(SELF_BOOTSTRAPPING_HEADER_OFFSET); + return headerBuffer; + } + + private static void commitChronicleMapReady( + VanillaChronicleHash map, RandomAccessFile raf, ByteBuffer headerBuffer, int headerSize) + throws IOException { + FileChannel fileChannel = raf.getChannel(); + // see HCOLL-396 + map.msync(); + + //noinspection PointlessBitwiseExpression + headerBuffer.putInt(SIZE_WORD_OFFSET, READY | DATA | headerSize); + headerBuffer.clear().position(SIZE_WORD_OFFSET).limit(SIZE_WORD_OFFSET + 4); + writeFully(fileChannel, SIZE_WORD_OFFSET, headerBuffer); + } + @Override public ChronicleMapBuilder clone() { try { @@ -475,22 +577,6 @@ public final class ChronicleMapBuilder implements return this; } - private static void checkSizeIsStaticallyKnown(SerializationBuilder builder, String role) { - if (builder.sizeIsStaticallyKnown) { - throw new IllegalStateException("Size of " + builder.tClass + - " instances is constant and statically known, shouldn't be specified via " + - "average" + role + "Size() or average" + role + "() methods"); - } - } - - private static void checkAverageSize(double averageSize, String role) { - if (averageSize <= 0 || isNaN(averageSize) || - Double.isInfinite(averageSize)) { - throw new IllegalArgumentException("Average " + role + " size must be a positive, " + - "finite number"); - } - } - /** * Configures the constant number of bytes, taken by serialized form of values, put into maps, * created by this builder. This is done by providing the {@code sampleValue}, all values should @@ -562,16 +648,6 @@ public final class ChronicleMapBuilder implements return keyBuilder; } - static class EntrySizeInfo { - final double averageEntrySize; - final int worstAlignment; - - public EntrySizeInfo(double averageEntrySize, int worstAlignment) { - this.averageEntrySize = averageEntrySize; - this.worstAlignment = worstAlignment; - } - } - private EntrySizeInfo entrySizeInfo() { double size = 0; double keySize = averageKeySize(); @@ -643,20 +719,6 @@ public final class ChronicleMapBuilder implements return keyBuilder.constantSizeMarshaller() || sampleKey != null; } - private static double averageSizeStoringLength( - SerializationBuilder builder, double averageSize) { - SizeMarshaller sizeMarshaller = builder.sizeMarshaller(); - if (averageSize == round(averageSize)) - return sizeMarshaller.storingLength(round(averageSize)); - long lower = roundDown(averageSize); - long upper = lower + 1; - int lowerStoringLength = sizeMarshaller.storingLength(lower); - int upperStoringLength = sizeMarshaller.storingLength(upper); - if (lowerStoringLength == upperStoringLength) - return lowerStoringLength; - return lower * (upper - averageSize) + upper * (averageSize - lower); - } - private int worstAlignmentAssumingChunkSize( long constantSizeBeforeAlignment, int chunkSize) { int alignment = valueAlignment(); @@ -687,11 +749,6 @@ public final class ChronicleMapBuilder implements this.worstAlignment = worstAlignment; } - static int greatestCommonDivisor(int a, int b) { - if (b == 0) return a; - return greatestCommonDivisor(b, a % b); - } - long chunkSize() { if (actualChunkSize > 0) return actualChunkSize; @@ -720,13 +777,6 @@ public final class ChronicleMapBuilder implements return (entrySizeInfo().averageEntrySize + chunkSize - 1) / chunkSize; } - private static int maxDefaultChunksPerAverageEntry(boolean replicated) { - // When replicated, having 8 chunks (=> 8 bits in bitsets) per entry seems more wasteful - // because when replicated we have bit sets per each remote node, not only allocation - // bit set as when non-replicated - return replicated ? 4 : 8; - } - @Override public ChronicleMapBuilder maxChunksPerEntry(int maxChunksPerEntry) { if (maxChunksPerEntry < 1) @@ -930,22 +980,6 @@ public final class ChronicleMapBuilder implements : segmentsForEntries; } - private static int estimateSegmentsForEntries(long size) { - if (size > 200 << 20) - return 256; - if (size >= 1 << 20) - return 128; - if (size >= 128 << 10) - return 64; - if (size >= 16 << 10) - return 32; - if (size >= 4 << 10) - return 16; - if (size >= 1 << 10) - return 8; - return 1; - } - @Override public ChronicleMapBuilder actualSegments(int actualSegments) { checkSegments(actualSegments); @@ -1501,7 +1535,7 @@ public final class ChronicleMapBuilder implements // truncated between length() and read() calls, then continue to wait } try { - Jvm.pause(100); + Thread.sleep(100); } catch (InterruptedException e) { if (recover) { break; @@ -1562,10 +1596,6 @@ public final class ChronicleMapBuilder implements return headerBuffer; } - private static long headerChecksum(ByteBuffer headerBuffer, int headerSize) { - return LongHashFunction.xx_r39().hashBytes(headerBuffer, SIZE_WORD_OFFSET, headerSize + 4); - } - private VanillaChronicleMap createWithNewFile( VanillaChronicleMap map, File file, RandomAccessFile raf, ByteBuffer headerBuffer, int headerSize) throws IOException { @@ -1625,63 +1655,6 @@ public final class ChronicleMapBuilder implements } } - private static void writeNotComplete( - FileChannel fileChannel, ByteBuffer headerBuffer, int headerSize) throws IOException { - //noinspection PointlessBitwiseExpression - headerBuffer.putInt(SIZE_WORD_OFFSET, NOT_COMPLETE | DATA | headerSize); - headerBuffer.clear().position(SIZE_WORD_OFFSET).limit(SIZE_WORD_OFFSET + 4); - writeFully(fileChannel, SIZE_WORD_OFFSET, headerBuffer); - } - - /** - * @return ByteBuffer, with self bootstrapping header in [position, limit) range - */ - private static ByteBuffer writeHeader( - FileChannel fileChannel, VanillaChronicleMap map) throws IOException { - ByteBuffer headerBuffer = ByteBuffer.allocate( - SELF_BOOTSTRAPPING_HEADER_OFFSET + MAX_BOOTSTRAPPING_HEADER_SIZE); - headerBuffer.order(LITTLE_ENDIAN); - - Bytes headerBytes = Bytes.wrapForWrite(headerBuffer); - headerBytes.writePosition(SELF_BOOTSTRAPPING_HEADER_OFFSET); - Wire wire = new TextWire(headerBytes); - wire.getValueOut().typedMarshallable(map); - - int headerLimit = (int) headerBytes.writePosition(); - int headerSize = headerLimit - SELF_BOOTSTRAPPING_HEADER_OFFSET; - // First set readiness bit to READY, to compute checksum correctly - //noinspection PointlessBitwiseExpression - headerBuffer.putInt(SIZE_WORD_OFFSET, READY | DATA | headerSize); - - long checksum = headerChecksum(headerBuffer, headerSize); - headerBuffer.putLong(HEADER_OFFSET, checksum); - - // Set readiness bit to NOT_COMPLETE, because the Chronicle Map instance is not actually - // ready yet - //noinspection PointlessBitwiseExpression - headerBuffer.putInt(SIZE_WORD_OFFSET, NOT_COMPLETE | DATA | headerSize); - - // Write the size-prefixed blob to the file - headerBuffer.position(0).limit(headerLimit); - writeFully(fileChannel, 0, headerBuffer); - - headerBuffer.position(SELF_BOOTSTRAPPING_HEADER_OFFSET); - return headerBuffer; - } - - private static void commitChronicleMapReady( - VanillaChronicleHash map, RandomAccessFile raf, ByteBuffer headerBuffer, int headerSize) - throws IOException { - FileChannel fileChannel = raf.getChannel(); - // see HCOLL-396 - map.msync(); - - //noinspection PointlessBitwiseExpression - headerBuffer.putInt(SIZE_WORD_OFFSET, READY | DATA | headerSize); - headerBuffer.clear().position(SIZE_WORD_OFFSET).limit(SIZE_WORD_OFFSET + 4); - writeFully(fileChannel, SIZE_WORD_OFFSET, headerBuffer); - } - ChronicleMap createWithoutFile() { replicated = replicationIdentifier != -1; persisted = false; @@ -1778,10 +1751,10 @@ public final class ChronicleMapBuilder implements /** * Inject your SPI code around basic {@code ChronicleMap}'s operations with entries: * removing entries, replacing entries' value and inserting new entries. - * + * *

This affects behaviour of ordinary map.put(), map.remove(), etc. calls, as well as removes * and replacing values during iterations, remote map calls and - * internal replication operations. + * internal replication operations. */ public ChronicleMapBuilder entryOperations(MapEntryOperations entryOperations) { Objects.requireNonNull(entryOperations); @@ -1793,7 +1766,7 @@ public final class ChronicleMapBuilder implements * Inject your SPI around logic of all {@code ChronicleMap}'s operations with individual keys: * from {@link ChronicleMap#containsKey} to {@link ChronicleMap#acquireUsing} and * {@link ChronicleMap#merge}. - * + * *

This affects behaviour of ordinary map calls, as well as remote calls. */ public ChronicleMapBuilder mapMethods(MapMethods mapMethods) { @@ -1808,5 +1781,21 @@ public final class ChronicleMapBuilder implements this.remoteOperations = remoteOperations; return this; } + + enum ChecksumEntries {YES, NO, IF_PERSISTED} + + interface FileIOAction { + void fileIOAction() throws IOException; + } + + static class EntrySizeInfo { + final double averageEntrySize; + final int worstAlignment; + + public EntrySizeInfo(double averageEntrySize, int worstAlignment) { + this.averageEntrySize = averageEntrySize; + this.worstAlignment = worstAlignment; + } + } } diff --git a/src/test/java/net/openhft/chronicle/map/CHMLatencyTestMain.java b/src/test/java/net/openhft/chronicle/map/CHMLatencyTestMain.java index a52653577..a95d94c1d 100755 --- a/src/test/java/net/openhft/chronicle/map/CHMLatencyTestMain.java +++ b/src/test/java/net/openhft/chronicle/map/CHMLatencyTestMain.java @@ -17,6 +17,7 @@ package net.openhft.chronicle.map; import net.openhft.affinity.AffinityLock; +import net.openhft.chronicle.core.Jvm; import net.openhft.chronicle.core.values.LongValue; import net.openhft.chronicle.values.Values; @@ -75,8 +76,8 @@ public class CHMLatencyTestMain { file.delete(); ChronicleMap countersMap = ChronicleMapBuilder.of(LongValue.class, LongValue.class) - .entries(KEYS) - .createPersistedTo(file); + .entries(KEYS) + .createPersistedTo(file); // add keys LongValue key = Values.newHeapInstance(LongValue.class); @@ -142,13 +143,8 @@ public class CHMLatencyTestMain { @Override public void run() { - while (running) { - try { - Jvm.pause(1); - } catch (InterruptedException e) { - e.printStackTrace(); - break; - } + while (running && !Thread.currentThread().isInterrupted()) { + Jvm.pause(1); long delay = System.nanoTime() - sample; if (delay > 1000 * 1000) { System.out.println("\n" + (System.currentTimeMillis() - START_TIME) + " : Delay of " + delay / 100000 / 10.0 + " ms."); diff --git a/src/test/java/net/openhft/chronicle/map/CHMTest5.java b/src/test/java/net/openhft/chronicle/map/CHMTest5.java index d17a24237..e22e02b22 100644 --- a/src/test/java/net/openhft/chronicle/map/CHMTest5.java +++ b/src/test/java/net/openhft/chronicle/map/CHMTest5.java @@ -16,12 +16,12 @@ package net.openhft.chronicle.map; -import net.openhft.chronicle.algo.bytes.Access; import net.openhft.chronicle.algo.locks.AcquisitionStrategies; import net.openhft.chronicle.algo.locks.ReadWriteLockingStrategy; import net.openhft.chronicle.algo.locks.TryAcquireOperations; import net.openhft.chronicle.algo.locks.VanillaReadWriteWithWaitsLockingStrategy; import net.openhft.chronicle.bytes.Byteable; +import net.openhft.chronicle.map.utility.ProcessInstanceLimiter; import net.openhft.chronicle.values.Array; import net.openhft.chronicle.values.Group; import net.openhft.chronicle.values.Values; @@ -229,14 +229,7 @@ public class CHMTest5 { } public static void pause(long pause) { - long start = System.currentTimeMillis(); - long elapsedTime; - while ((elapsedTime = System.currentTimeMillis() - start) < pause) { - try { - Jvm.pause(pause - elapsedTime); - } catch (InterruptedException e) { - } - } + ProcessInstanceLimiter.pause(pause); } public interface CHMTest5Data { diff --git a/src/test/java/net/openhft/chronicle/map/CHMUseCasesTest.java b/src/test/java/net/openhft/chronicle/map/CHMUseCasesTest.java index 8279de769..94db8d75c 100644 --- a/src/test/java/net/openhft/chronicle/map/CHMUseCasesTest.java +++ b/src/test/java/net/openhft/chronicle/map/CHMUseCasesTest.java @@ -19,7 +19,7 @@ package net.openhft.chronicle.map; import com.google.common.primitives.Chars; import net.openhft.chronicle.bytes.Bytes; import net.openhft.chronicle.bytes.BytesMarshallable; -import net.openhft.chronicle.core.Maths; +import net.openhft.chronicle.core.Jvm; import net.openhft.chronicle.core.util.SerializableFunction; import net.openhft.chronicle.core.values.*; import net.openhft.chronicle.hash.serialization.DataAccess; @@ -54,6 +54,103 @@ import static net.openhft.chronicle.core.Maths.divideRoundUp; import static net.openhft.chronicle.map.fromdocs.OpenJDKAndHashMapExamplesTest.parseYYYYMMDD; import static org.junit.Assert.*; +enum ToString implements SerializableFunction { + INSTANCE; + + @Override + public String apply(Object o) { + return String.valueOf(o); + } + +} + +interface IData extends BytesMarshallable { + String getText(); + + void setText(@MaxUtf8Length(64) String text); + + int getNumber(); + + void setNumber(int number); + + class Data implements IData, BytesMarshallable { + static final long MAGIC = 0x8081828384858687L; + static final long MAGIC2 = 0xa0a1a2a3a4a5a6a7L; + + String text; + int number; + + @Override + public String getText() { + return text; + } + + @Override + public void setText(String text) { + this.text = text; + } + + @Override + public int getNumber() { + return number; + } + + @Override + public void setNumber(int number) { + this.number = number; + } + + @Override + public void readMarshallable(@NotNull Bytes in) throws IllegalStateException { + long magic = in.readLong(); + if (magic != MAGIC) + throw new AssertionError("Start " + Long.toHexString(magic)); + text = in.readUTFΔ(); + number = in.readInt(); + long magic2 = in.readLong(); + if (magic2 != MAGIC2) + throw new AssertionError("End " + Long.toHexString(magic2)); + } + + @Override + public void writeMarshallable(@NotNull Bytes out) { + out.writeLong(MAGIC); + out.writeUTFΔ(text); + out.writeInt(number); + out.writeLong(MAGIC2); + } + } +} + +interface IBean { + long getLong(); + + void setLong(long num); + + double getDouble(); + + void setDouble(double d); + + int getInt(); + + void setInt(int i); + + @Array(length = 7) + void setInnerAt(int index, Inner inner); + + Inner getInnerAt(int index); + + /* nested interface - empowering an Off-Heap hierarchical “TIER of prices” + as array[ ] value */ + interface Inner { + + String getMessage(); + + void setMessage(@MaxUtf8Length(20) String px); + + } +} + /** * This test enumerates common use cases for keys and values. */ @@ -61,16 +158,47 @@ import static org.junit.Assert.*; public class CHMUseCasesTest { - enum TypeOfMap {SIMPLE, SIMPLE_PERSISTED} - + static volatile int i = 0; private final TypeOfMap typeOfMap; Collection closeables = new ArrayList(); + ChronicleMap map1; + ChronicleMap map2; public CHMUseCasesTest(TypeOfMap typeOfMap) { this.typeOfMap = typeOfMap; } + @Parameterized.Parameters + public static Collection data() { + return asList(new Object[][]{ + { + TypeOfMap.SIMPLE + }, + + { + TypeOfMap.SIMPLE_PERSISTED + } + }); + + } + + private static void appendMode(ByteBuffer valueA) { + valueA.position(valueA.limit()); + valueA.limit(valueA.capacity()); + } + + public static Map mapOf(K k, V v, Object... keysAndValues) { + Map ret = new LinkedHashMap<>(); + ret.put(k, v); + for (int i = 0; i < keysAndValues.length - 1; i += 2) { + Object key = keysAndValues[i]; + Object value = keysAndValues[i + 1]; + ret.put((K) key, (V) value); + } + return ret; + } + @After public void after() { for (Closeable c : closeables) { @@ -86,24 +214,6 @@ public class CHMUseCasesTest { map1 = null; } - ChronicleMap map1; - ChronicleMap map2; - - @Parameterized.Parameters - public static Collection data() { - return asList(new Object[][]{ - { - TypeOfMap.SIMPLE - }, - - { - TypeOfMap.SIMPLE_PERSISTED - } - }); - - } - - /** * * waits until map1 and map2 show the same value * @@ -114,15 +224,10 @@ public class CHMUseCasesTest { for (; t < timeOutMs; t++) { if (map1.equals(map2)) break; - try { Jvm.pause(1); - } catch (InterruptedException e) { - throw new RuntimeException(e); - } } } - private void mapChecks() { if (typeOfMap == TypeOfMap.SIMPLE) checkJsonSerialization(); @@ -155,7 +260,6 @@ public class CHMUseCasesTest { } } - private void checkJsonSerialization() { File file = null; try { @@ -201,8 +305,6 @@ public class CHMUseCasesTest { } } - static volatile int i = 0; - private ChronicleMap newInstance(ChronicleMapBuilder builder) throws IOException { if (!builder.constantlySizedKeys() && builder.averageKey == null) @@ -238,38 +340,6 @@ public class CHMUseCasesTest { } } - static class PrefixStringFunction implements SerializableFunction { - private final String prefix; - - public PrefixStringFunction(@NotNull String prefix) { - this.prefix = prefix; - } - - @Override - public String apply(String s) { - return prefix + s; - } - - @Override - public boolean equals(Object obj) { - return obj instanceof PrefixStringFunction && - prefix.equals(((PrefixStringFunction) obj).prefix); - } - - @Override - public String toString() { - return prefix; - } - } - - - - interface I1 { - @Array(length = 10) - String getStrAt(int i); - void setStrAt(int i, @MaxUtf8Length(10) String str); - } - @Test public void testArrayOfString() throws ExecutionException, InterruptedException, IOException { @@ -313,7 +383,6 @@ public class CHMUseCasesTest { } } - @Test public void testCharArrayValue() throws ExecutionException, InterruptedException, IOException { @@ -401,7 +470,6 @@ public class CHMUseCasesTest { } } - @Test public void testEntrySpanningSeveralChunks() throws ExecutionException, InterruptedException, IOException { @@ -424,10 +492,6 @@ public class CHMUseCasesTest { } } - - - - @Test public void testKeyValueSizeBySample() throws ExecutionException, InterruptedException, IOException { @@ -444,7 +508,6 @@ public class CHMUseCasesTest { } } - @Test public void testLargeCharSequenceValue() throws ExecutionException, InterruptedException, IOException { @@ -483,23 +546,6 @@ public class CHMUseCasesTest { } } - - private static class StringPrefixUnaryOperator - implements BiFunction, Serializable { - - private String prefix; - - StringPrefixUnaryOperator(final String prefix1) { - prefix = prefix1; - } - - @Override - public String apply(String k, String v) { - return prefix + v; - } - } - - @Test public void testStringStringMapMutableValue() throws ExecutionException, InterruptedException, IOException { @@ -548,7 +594,6 @@ public class CHMUseCasesTest { } } - /** * CharSequence is more efficient when object creation is avoided. * * The key can only be on heap and variable length serialised. @@ -598,7 +643,6 @@ public class CHMUseCasesTest { } } - @Test public void testAcquireUsingWithCharSequence() throws IOException { @@ -716,12 +760,6 @@ public class CHMUseCasesTest { } } - interface StringValue { - CharSequence getValue(); - void getUsingValue(StringBuilder using); - void setValue(@net.openhft.chronicle.values.NotNull @MaxUtf8Length(64) CharSequence value); - } - /** * StringValue represents any bean which contains a String Value */ @@ -1050,7 +1088,6 @@ public class CHMUseCasesTest { } } - @Test public void testByteBufferByteBufferDefaultKeyValueMarshaller() throws ExecutionException, InterruptedException, IOException { @@ -1080,7 +1117,6 @@ public class CHMUseCasesTest { } } - @Test public void testByteBufferByteBufferMap() throws ExecutionException, InterruptedException, IOException { @@ -1193,11 +1229,6 @@ public class CHMUseCasesTest { } } - private static void appendMode(ByteBuffer valueA) { - valueA.position(valueA.limit()); - valueA.limit(valueA.capacity()); - } - @Test public void testByteBufferDirectByteBufferMap() throws ExecutionException, InterruptedException, IOException { @@ -1353,12 +1384,6 @@ public class CHMUseCasesTest { } } - interface UnsignedIntValue { - long getValue(); - void setValue(@Range(min = 0, max = (1L << 32) - 1) long value); - long addValue(long addition); - } - /** * For unsigned int -> unsigned int entries, the key can be on heap or off heap. */ @@ -1583,12 +1608,6 @@ public class CHMUseCasesTest { } } - interface UnsignedShortValue { - int getValue(); - void setValue(@Range(min = 0, max = Character.MAX_VALUE) int value); - int addValue(int addition); - } - /** * For int -> unsigned short values, the key can be on heap or off heap. */ @@ -1794,12 +1813,6 @@ public class CHMUseCasesTest { } } - interface UnsignedByteValue { - int getValue(); - void setValue(@Range(min = 0, max = 255) int value); - int addValue(int addition); - } - /** * For int-> byte entries, the key can be on heap or off heap. */ @@ -2114,7 +2127,6 @@ public class CHMUseCasesTest { } } - /** * For double values, the key can be on heap or off heap. */ @@ -2422,7 +2434,6 @@ public class CHMUseCasesTest { } } - @Test public void testMapStringStringValue() throws IOException { @@ -2523,17 +2534,6 @@ public class CHMUseCasesTest { } } - public static Map mapOf(K k, V v, Object... keysAndValues) { - Map ret = new LinkedHashMap<>(); - ret.put(k, v); - for (int i = 0; i < keysAndValues.length - 1; i += 2) { - Object key = keysAndValues[i]; - Object value = keysAndValues[i + 1]; - ret.put((K) key, (V) value); - } - return ret; - } - @Test public void testGeneratedDataValue() throws IOException { @@ -2599,6 +2599,87 @@ public class CHMUseCasesTest { } } + enum TypeOfMap {SIMPLE, SIMPLE_PERSISTED} + + + interface I1 { + @Array(length = 10) + String getStrAt(int i); + + void setStrAt(int i, @MaxUtf8Length(10) String str); + } + + interface StringValue { + CharSequence getValue(); + + void setValue(@net.openhft.chronicle.values.NotNull @MaxUtf8Length(64) CharSequence value); + + void getUsingValue(StringBuilder using); + } + + interface UnsignedIntValue { + long getValue(); + + void setValue(@Range(min = 0, max = (1L << 32) - 1) long value); + + long addValue(long addition); + } + + interface UnsignedShortValue { + int getValue(); + + void setValue(@Range(min = 0, max = Character.MAX_VALUE) int value); + + int addValue(int addition); + } + + interface UnsignedByteValue { + int getValue(); + + void setValue(@Range(min = 0, max = 255) int value); + + int addValue(int addition); + } + + static class PrefixStringFunction implements SerializableFunction { + private final String prefix; + + public PrefixStringFunction(@NotNull String prefix) { + this.prefix = prefix; + } + + @Override + public String apply(String s) { + return prefix + s; + } + + @Override + public boolean equals(Object obj) { + return obj instanceof PrefixStringFunction && + prefix.equals(((PrefixStringFunction) obj).prefix); + } + + @Override + public String toString() { + return prefix; + } + } + + private static class StringPrefixUnaryOperator + implements BiFunction, Serializable { + + private String prefix; + + StringPrefixUnaryOperator(final String prefix1) { + prefix = prefix1; + } + + @Override + public String apply(String k, String v) { + return prefix + v; + } + } + private static class DataDataAccess extends BytesMarshallableDataAccess { public DataDataAccess() { super(IData.Data.class); @@ -2626,100 +2707,3 @@ public class CHMUseCasesTest { } } } - -interface IData extends BytesMarshallable { - void setText(@MaxUtf8Length(64) String text); - - void setNumber(int number); - - String getText(); - - int getNumber(); - - class Data implements IData, BytesMarshallable { - static final long MAGIC = 0x8081828384858687L; - static final long MAGIC2 = 0xa0a1a2a3a4a5a6a7L; - - String text; - int number; - - @Override - public String getText() { - return text; - } - - @Override - public void setText(String text) { - this.text = text; - } - - @Override - public int getNumber() { - return number; - } - - @Override - public void setNumber(int number) { - this.number = number; - } - - @Override - public void readMarshallable(@NotNull Bytes in) throws IllegalStateException { - long magic = in.readLong(); - if (magic != MAGIC) - throw new AssertionError("Start " + Long.toHexString(magic)); - text = in.readUTFΔ(); - number = in.readInt(); - long magic2 = in.readLong(); - if (magic2 != MAGIC2) - throw new AssertionError("End " + Long.toHexString(magic2)); - } - - @Override - public void writeMarshallable(@NotNull Bytes out) { - out.writeLong(MAGIC); - out.writeUTFΔ(text); - out.writeInt(number); - out.writeLong(MAGIC2); - } - } -} - -enum ToString implements SerializableFunction { - INSTANCE; - - @Override - public String apply(Object o) { - return String.valueOf(o); - } - -} - -interface IBean { - long getLong(); - - void setLong(long num); - - double getDouble(); - - void setDouble(double d); - - int getInt(); - - void setInt(int i); - - @Array(length = 7) - void setInnerAt(int index, Inner inner); - - Inner getInnerAt(int index); - - /* nested interface - empowering an Off-Heap hierarchical “TIER of prices” - as array[ ] value */ - interface Inner { - - String getMessage(); - - void setMessage(@MaxUtf8Length(20) String px); - - } -} diff --git a/src/test/java/net/openhft/chronicle/map/ChronicleMapSanityCheckTest.java b/src/test/java/net/openhft/chronicle/map/ChronicleMapSanityCheckTest.java index 6339b343e..e4c0688d2 100644 --- a/src/test/java/net/openhft/chronicle/map/ChronicleMapSanityCheckTest.java +++ b/src/test/java/net/openhft/chronicle/map/ChronicleMapSanityCheckTest.java @@ -16,6 +16,7 @@ package net.openhft.chronicle.map; +import net.openhft.chronicle.core.Jvm; import org.junit.Test; import java.io.File; @@ -33,10 +34,6 @@ import java.util.concurrent.locks.LockSupport; */ public class ChronicleMapSanityCheckTest { - enum DummyValue { - DUMMY_VALUE - } - @Test public void testSanity1() throws IOException, InterruptedException { @@ -124,4 +121,8 @@ public class ChronicleMapSanityCheckTest { } } + + enum DummyValue { + DUMMY_VALUE + } } diff --git a/src/test/java/net/openhft/chronicle/map/Issue62ChronicleClient.java b/src/test/java/net/openhft/chronicle/map/Issue62ChronicleClient.java index a5abd1be6..f8a604b9f 100644 --- a/src/test/java/net/openhft/chronicle/map/Issue62ChronicleClient.java +++ b/src/test/java/net/openhft/chronicle/map/Issue62ChronicleClient.java @@ -16,6 +16,7 @@ package net.openhft.chronicle.map; +import net.openhft.chronicle.core.Jvm; import net.openhft.chronicle.hash.ChronicleHashBuilderPrivateAPI; import org.slf4j.Logger; import org.slf4j.LoggerFactory; diff --git a/src/test/java/net/openhft/chronicle/map/Issue62ChronicleServer.java b/src/test/java/net/openhft/chronicle/map/Issue62ChronicleServer.java index bf8818bd9..1b6f806e2 100644 --- a/src/test/java/net/openhft/chronicle/map/Issue62ChronicleServer.java +++ b/src/test/java/net/openhft/chronicle/map/Issue62ChronicleServer.java @@ -16,6 +16,7 @@ package net.openhft.chronicle.map; +import net.openhft.chronicle.core.Jvm; import net.openhft.chronicle.hash.ChronicleHashBuilderPrivateAPI; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -25,16 +26,14 @@ import java.nio.file.Paths; public class Issue62ChronicleServer { - private final static Logger LOGGER = LoggerFactory.getLogger(Issue62ChronicleServer.class); - final static String STR = //"This is just a long string, which causes sink to fail for some reason."; "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"; - - private final static File MAP_FILE_A = - Paths.get(System.getProperty("java.io.tmpdir"), "map.a").toFile(); final static File MAP_FILE_B = Paths.get(System.getProperty("java.io.tmpdir"), "map.b").toFile(); + private final static Logger LOGGER = LoggerFactory.getLogger(Issue62ChronicleServer.class); + private final static File MAP_FILE_A = + Paths.get(System.getProperty("java.io.tmpdir"), "map.a").toFile(); static void prepare(File file) { if (file.exists()) diff --git a/src/test/java/net/openhft/chronicle/map/Issue63Test.java b/src/test/java/net/openhft/chronicle/map/Issue63Test.java index 4fbad3478..b26271979 100644 --- a/src/test/java/net/openhft/chronicle/map/Issue63Test.java +++ b/src/test/java/net/openhft/chronicle/map/Issue63Test.java @@ -24,6 +24,7 @@ import net.openhft.chronicle.set.*; import org.junit.Test; import java.io.File; +import java.io.IOException; import java.nio.file.Files; import java.nio.file.Path; import java.nio.file.Paths; @@ -38,8 +39,54 @@ import static org.junit.Assert.assertEquals; public class Issue63Test { + // Right now no corresponding "knownUsers" object + private ChronicleMap> knownItems; + private ChronicleMap xVectors; + private ChronicleSet xRecentIDs; + private ChronicleMap yVectors; + private ChronicleSet yRecentIDs; + + public static void main(String[] args) throws Exception { + new Issue63Test().testChronicleMap(); + } + + private static void putIntoMapAndRecentSet( + ChronicleMap map, ChronicleSet recentSet, + String key, float[] value) { + try (ExternalMapQueryContext c = map.queryContext(key); + ExternalSetQueryContext setC = recentSet.queryContext(key)) { + if (c.writeLock().tryLock(1, MINUTES) && setC.writeLock().tryLock(1, MINUTES)) { + putNoReturn(c, value); + addNoReturn(setC); + } else { + throw new RuntimeException("Dead lock"); + } + } catch (InterruptedException e) { + throw new RuntimeException(e); + } + } + + private static void addNoReturn(SetQueryContext c) { + SetAbsentEntry setAbsentEntry = c.absentEntry(); + if (setAbsentEntry != null) { + c.insert(setAbsentEntry); + } + } + + private static void putNoReturn(MapQueryContext c, V value) { + MapEntry entry = c.entry(); + Data newValue = c.wrapValueAsData(value); + if (entry != null) { + c.replaceValue(entry, newValue); + } else { + MapAbsentEntry absentEntry = c.absentEntry(); + assert absentEntry != null; + c.insert(absentEntry, newValue); + } + } + @Test - public void issue63test() { + public void issue63test() throws IOException { Path path = Paths.get(System.getProperty("java.io.tmpdir") + "/test-vectors1.dat"); if (Files.exists(path)) Files.delete(path); File mapFile = path.toFile(); @@ -137,18 +184,7 @@ public class Issue63Test { assertArrayEquals(exp2, xVectors2.get(key2), 0.0f); } - // Right now no corresponding "knownUsers" object - private ChronicleMap> knownItems; - private ChronicleMap xVectors; - private ChronicleSet xRecentIDs; - private ChronicleMap yVectors; - private ChronicleSet yRecentIDs; - - public static void main(String[] args) throws Exception { - new Issue63Test().testChronicleMap(); - } - - void testChronicleMap() { + void testChronicleMap() throws IOException { int num = 1_000_000; testChronicleMap(System.getProperty("java.io.tmpdir"), num, num); ThreadLocalRandom random = ThreadLocalRandom.current(); @@ -169,8 +205,7 @@ public class Issue63Test { void testChronicleMap(String persistToDir, int numXIDs, - int numYIDs) - throws Exception { + int numYIDs) throws IOException { if (!Files.exists(Paths.get(persistToDir))) Files.createDirectory(Paths.get(persistToDir)); @@ -257,41 +292,6 @@ public class Issue63Test { putIntoMapAndRecentSet(yVectors, yRecentIDs, id, arr); } - private static void putIntoMapAndRecentSet( - ChronicleMap map, ChronicleSet recentSet, - String key, float[] value) { - try (ExternalMapQueryContext c = map.queryContext(key); - ExternalSetQueryContext setC = recentSet.queryContext(key)) { - if (c.writeLock().tryLock(1, MINUTES) && setC.writeLock().tryLock(1, MINUTES)) { - putNoReturn(c, value); - addNoReturn(setC); - } else { - throw new RuntimeException("Dead lock"); - } - } catch (InterruptedException e) { - throw new RuntimeException(e); - } - } - - private static void addNoReturn(SetQueryContext c) { - SetAbsentEntry setAbsentEntry = c.absentEntry(); - if (setAbsentEntry != null) { - c.insert(setAbsentEntry); - } - } - - private static void putNoReturn(MapQueryContext c, V value) { - MapEntry entry = c.entry(); - Data newValue = c.wrapValueAsData(value); - if (entry != null) { - c.replaceValue(entry, newValue); - } else { - MapAbsentEntry absentEntry = c.absentEntry(); - assert absentEntry != null; - c.insert(absentEntry, newValue); - } - } - public void addKnownItems(ChronicleMap> knownItems, String id, List items) { try (ExternalMapQueryContext, ?> c = @@ -307,7 +307,7 @@ public class Issue63Test { } @Test - public void testKnownItems() { + public void testKnownItems() throws IOException { ArrayList averageKnownItems = new ArrayList<>(); for (int i = 0; i < 100; i++) { diff --git a/src/test/java/net/openhft/chronicle/map/ProcessInstanceLimiterMain.java b/src/test/java/net/openhft/chronicle/map/ProcessInstanceLimiterMain.java index 6977e8f8b..6d30e8ac1 100644 --- a/src/test/java/net/openhft/chronicle/map/ProcessInstanceLimiterMain.java +++ b/src/test/java/net/openhft/chronicle/map/ProcessInstanceLimiterMain.java @@ -16,6 +16,9 @@ package net.openhft.chronicle.map; +import net.openhft.chronicle.core.Jvm; +import net.openhft.chronicle.map.utility.ProcessInstanceLimiter; + import java.io.File; import java.io.IOException; import java.io.Serializable; @@ -64,14 +67,7 @@ public class ProcessInstanceLimiterMain implements Runnable { } public static void pause(long pause) { - long start = System.currentTimeMillis(); - long elapsedTime; - while ((elapsedTime = System.currentTimeMillis() - start) < pause) { - try { - Jvm.pause(pause - elapsedTime); - } catch (InterruptedException e) { - } - } + ProcessInstanceLimiter.pause(pause); } public void run() { diff --git a/src/test/java/net/openhft/chronicle/map/ReplicationCheckingMap.java b/src/test/java/net/openhft/chronicle/map/ReplicationCheckingMap.java index 3642e8f2d..a15c868a1 100644 --- a/src/test/java/net/openhft/chronicle/map/ReplicationCheckingMap.java +++ b/src/test/java/net/openhft/chronicle/map/ReplicationCheckingMap.java @@ -17,6 +17,7 @@ package net.openhft.chronicle.map; import net.openhft.chronicle.bytes.BytesStore; +import net.openhft.chronicle.core.Jvm; import net.openhft.chronicle.core.io.Closeable; import net.openhft.chronicle.core.util.SerializableFunction; import org.jetbrains.annotations.NotNull; @@ -91,13 +92,10 @@ public class ReplicationCheckingMap implements ChronicleMap { return r1; if (i > 30) { - try { Jvm.pause(i); - } catch (InterruptedException e) { - e.printStackTrace(); - } - } else + } else { Thread.yield(); + } } Assert.assertEquals(map1, map2); diff --git a/src/test/java/net/openhft/chronicle/map/utility/ProcessInstanceLimiter.java b/src/test/java/net/openhft/chronicle/map/utility/ProcessInstanceLimiter.java index 7f6f3fe88..14bb8da01 100644 --- a/src/test/java/net/openhft/chronicle/map/utility/ProcessInstanceLimiter.java +++ b/src/test/java/net/openhft/chronicle/map/utility/ProcessInstanceLimiter.java @@ -17,9 +17,12 @@ package net.openhft.chronicle.map.utility; import net.openhft.affinity.AffinitySupport; -import net.openhft.chronicle.algo.bytes.Access; -import net.openhft.chronicle.algo.locks.*; +import net.openhft.chronicle.algo.locks.AcquisitionStrategies; +import net.openhft.chronicle.algo.locks.ReadWriteLockingStrategy; +import net.openhft.chronicle.algo.locks.TryAcquireOperations; +import net.openhft.chronicle.algo.locks.VanillaReadWriteWithWaitsLockingStrategy; import net.openhft.chronicle.bytes.Byteable; +import net.openhft.chronicle.core.Jvm; import net.openhft.chronicle.map.ChronicleMap; import net.openhft.chronicle.map.ChronicleMapBuilder; import net.openhft.chronicle.values.Array; @@ -43,13 +46,13 @@ import static net.openhft.chronicle.algo.bytes.Access.checkedBytesStoreAccess; * shared map (using ChronicleMap) to maintain shared data across any processes * which use a ProcessInstanceLimiter, and checking on startup and regularly * after whether it is allowed to run. - * + *

*

Typically, you need to specify two things to create an instance of * ProcessInstanceLimiter: a path to a file that will hold the shared map; and a * callback object (an instance implementing ProcessInstanceLimiter.Callback) to * handle the various possible callback messages that the ProcessInstanceLimiter * can generate. - * + *

*

Once you have a ProcessInstanceLimiter instance, you specify a type of * process (any string) which will be limited to up to N processes running at * the same time by calling the setMaxNumberOfProcessesOfType() method. Finally @@ -57,18 +60,18 @@ import static net.openhft.chronicle.algo.bytes.Access.checkedBytesStoreAccess; * startingProcessOfType(X). This last is deliberately not done automatically as * you may wish for one type of process to define limitations on other types of * processes. - * + *

*

The are some convenience methods which allow you to quickly specify a limit * without consideration of the above. For example, if during your application * startup you call ProcessInstanceLimiter.limitTo(2), then you need not call * anything else and you have limited your application to running at most 2 JVM * instances of your application. Under the covers, this call is identical to * the sequence: - * + *

*

ProcessInstanceLimiter limiter = new ProcessInstanceLimiter(); * limiter.setMaxNumberOfProcessesOfType(processType,numProcesses); * limiter.startingProcessOfType(processType); - * + *

*

This: * 1. Creates a shared file called ProcessInstanceLimiter_DEFAULT_SHARED_MAP_ in * the temp directory to hold an instance of ChronicleMap @@ -83,7 +86,6 @@ import static net.openhft.chronicle.algo.bytes.Access.checkedBytesStoreAccess; */ public class ProcessInstanceLimiter implements Runnable { private static final long DEFAULT_TIME_UPDATE_INTERVAL_MS = 100L; - private long timeUpdateInterval = DEFAULT_TIME_UPDATE_INTERVAL_MS; private static final String DEFAULT_SHARED_MAP_NAME = "ProcessInstanceLimiter_DEFAULT_SHARED_MAP_"; private static final String DEFAULT_SHARED_MAP_DIRECTORY = System.getProperty("java.io.tmpdir"); private static final String DEFAULT_PROCESS_NAME = "_DEFAULT_"; @@ -97,6 +99,7 @@ public class ProcessInstanceLimiter implements Runnable { private final Callback callback; private final Map localUpdates = new ConcurrentHashMap(); private final Map processTypeToStartTimeType = new ConcurrentHashMap(); + private long timeUpdateInterval = DEFAULT_TIME_UPDATE_INTERVAL_MS; private long startTime; private long[] lastStartTimes; private Map timedata = new ConcurrentHashMap(); @@ -156,7 +159,7 @@ public class ProcessInstanceLimiter implements Runnable { /** * Convenience method. - * + *

*

Create a ProcessInstanceLimiter instance which is limited to one OS * process instance of the DEFAULT type. This will enforce that any JVM on * the same box which runs the code @@ -172,7 +175,7 @@ public class ProcessInstanceLimiter implements Runnable { /** * Convenience method. - * + *

*

Create a ProcessInstanceLimiter instance which is limited to * "numProcesses" OS process instances of the DEFAULT type. This will * enforce that any JVM on the same box which runs the code @@ -191,7 +194,7 @@ public class ProcessInstanceLimiter implements Runnable { /** * Convenience method. - * + *

*

Create a ProcessInstanceLimiter instance which is limited to * "numProcesses" OS process instances of the "processType" type. This will * enforce that any JVM on the same box which runs the code @@ -223,10 +226,8 @@ public class ProcessInstanceLimiter implements Runnable { long start = System.currentTimeMillis(); long elapsedTime; while ((elapsedTime = System.currentTimeMillis() - start) < pause) { - try { - Jvm.pause(pause - elapsedTime); - } catch (InterruptedException e) { - } + Thread.interrupted();// clear interruptions. + Jvm.pause(pause - elapsedTime); } } @@ -459,9 +460,9 @@ public class ProcessInstanceLimiter implements Runnable { private void unlock(Data data) { try { - VanillaReadWriteWithWaitsLockingStrategy.instance() - .writeUnlock(checkedBytesStoreAccess(), - ((Byteable) data).bytesStore(), ((Byteable) data).offset()); + VanillaReadWriteWithWaitsLockingStrategy.instance() + .writeUnlock(checkedBytesStoreAccess(), + ((Byteable) data).bytesStore(), ((Byteable) data).offset()); } catch (IllegalMonitorStateException e) { //odd, but we'll be unlocked either way System.out.println("Unexpected state: " + e); @@ -531,25 +532,30 @@ public class ProcessInstanceLimiter implements Runnable { /** * The Data object holds an array of timestamps and a maximum number of * processes allowed to be running concurrently - * + *

*

The Timelock field is just for locking the time field */ public interface Data { - /** Ensure lock goes first in flyweight layout, to apply external locking + /** + * Ensure lock goes first in flyweight layout, to apply external locking */ @Group(0) long getTimeLock(); + void setTimeLock(long timeLock); @Group(1) @Array(length = 50) void setTimeAt(int index, long time); + long getTimeAt(int index); @Group(1) int getMaxNumberOfProcessesAllowed(); + void setMaxNumberOfProcessesAllowed(int num); + boolean compareAndSwapMaxNumberOfProcessesAllowed(int expected, int value); } diff --git a/src/test/java/net/openhft/chronicle/set/Builder.java b/src/test/java/net/openhft/chronicle/set/Builder.java index 96dbf546e..19e5e9881 100644 --- a/src/test/java/net/openhft/chronicle/set/Builder.java +++ b/src/test/java/net/openhft/chronicle/set/Builder.java @@ -16,6 +16,8 @@ package net.openhft.chronicle.set; +import net.openhft.chronicle.core.Jvm; + import java.io.File; import java.io.IOException; import java.io.RandomAccessFile; @@ -52,7 +54,7 @@ public class Builder { return file; } - public static void waitTillEqual(Map map1, Map map2, int timeOutMs) throws InterruptedException { + public static void waitTillEqual(Map map1, Map map2, int timeOutMs) { int numberOfTimesTheSame = 0; long startTime = System.currentTimeMillis(); for (int t = 0; t < timeOutMs + 100; t++) {