From a956ac2a0e60bb0629c620f1aa8068ce183006c1 Mon Sep 17 00:00:00 2001 From: Peter Lawrey Date: Mon, 27 Oct 2014 15:59:22 +0000 Subject: [PATCH] Enhanced thread safety tests with the latest lang release --- .../impl/ByteableMarshaller.java | 14 ++++- .../openhft/chronicle/map/NodeDiscovery.java | 6 +- .../chronicle/map/ReplicatedChronicleMap.java | 26 ++++---- .../chronicle/map/StatelessChronicleMap.java | 2 +- .../openhft/chronicle/map/TcpReplicator.java | 4 +- .../chronicle/map/VanillaChronicleMap.java | 62 ++++++++++--------- .../map/ByteableIntValueMarshaller.java | 3 +- .../map/ByteableLongValueMarshaller.java | 3 +- .../chronicle/map/ChronicleMapTest.java | 12 ++-- .../chronicle/map/IntIntMultiMapTest.java | 20 +++++- .../map/TCPSocketReplicationIntValueTest.java | 12 ++-- ...CPSocketReplicationTest3VoidValueTest.java | 2 +- 12 files changed, 102 insertions(+), 64 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/common/serialization/impl/ByteableMarshaller.java b/src/main/java/net/openhft/chronicle/common/serialization/impl/ByteableMarshaller.java index fc2929e04..53080c63c 100644 --- a/src/main/java/net/openhft/chronicle/common/serialization/impl/ByteableMarshaller.java +++ b/src/main/java/net/openhft/chronicle/common/serialization/impl/ByteableMarshaller.java @@ -21,6 +21,7 @@ package net.openhft.chronicle.common.serialization.impl; import net.openhft.chronicle.common.serialization.AgileBytesMarshaller; import net.openhft.chronicle.common.serialization.Hasher; import net.openhft.lang.io.Bytes; +import net.openhft.lang.io.MultiStoreBytes; import net.openhft.lang.io.NativeBytes; import net.openhft.lang.io.serialization.*; import net.openhft.lang.io.serialization.impl.AllocateInstanceObjectFactory; @@ -117,7 +118,7 @@ public class ByteableMarshaller implements AgileBytesMarshal try { if (e == null) e = getInstance(); - e.bytes(bytes, bytes.position()); + setBytesAndOffset(e, bytes); bytes.skip(size); return e; } catch (Exception ex) { @@ -125,6 +126,17 @@ public class ByteableMarshaller implements AgileBytesMarshal } } + public static void setBytesAndOffset(Byteable e, Bytes bytes) { + // use the unshifting underlying Bytes (the position never changes) + // if you use a Bytes where the position changes, you risk concurrency issues. + if (bytes instanceof MultiStoreBytes) { + MultiStoreBytes msb = (MultiStoreBytes) bytes; + e.bytes(msb.underlyingBytes(), msb.underlyingOffset() + msb.position()); + } else { + e.bytes(bytes, bytes.position()); + } + } + @SuppressWarnings("unchecked") @NotNull E getInstance() throws Exception { diff --git a/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java b/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java index f3971b80f..9c3fa166e 100644 --- a/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java +++ b/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java @@ -259,9 +259,9 @@ public class NodeDiscovery { * bitwise OR's the two bit sets or put another way, merges the source bitset into the * destination bitset and returns the destination * - * @param source - * @param destination - * @return + * @param source to or + * @param destination to or + * @return result. */ private DirectBitSet orBitSets(@NotNull final DirectBitSet source, @NotNull final DirectBitSet destination) { diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java index 4fddeab91..7b1618f47 100644 --- a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java @@ -621,7 +621,7 @@ class ReplicatedChronicleMap, readLock(); try { long keySize = metaKeyWriter.size(keyWriter, key); - MultiStoreBytes entry = tmpBytes; + MultiStoreBytes entry = acquireTmpBytes(); long offset = searchKey(keyWriter, metaKeyWriter, key, keySize, hash2, entry, hashLookupLiveOnly); if (offset >= 0) { @@ -816,7 +816,7 @@ class ReplicatedChronicleMap, hashLookup.startSearch(hash2); for (long pos; (pos = hashLookup.nextPos()) >= 0L; ) { long offset = offsetFromPos(pos); - NativeBytes entry = entry(offset); + MultiStoreBytes entry = entry(offset); if (!keyEquals(keyWriter, metaKeyWriter, key, keySize, entry)) continue; // key is found @@ -982,7 +982,7 @@ class ReplicatedChronicleMap, multiMap.startSearch(hash2); for (long pos; (pos = multiMap.nextPos()) >= 0L; ) { long offset = offsetFromPos(pos); - NativeBytes entry = entry(offset); + MultiStoreBytes entry = entry(offset); if (!keyEquals(keyWriter, metaKeyWriter, key, keySize, entry)) continue; // key is found @@ -1049,7 +1049,7 @@ class ReplicatedChronicleMap, hashLookupLiveOnly.startSearch(hash2); for (long pos; (pos = hashLookupLiveOnly.nextPos()) >= 0L; ) { long offset = offsetFromPos(pos); - NativeBytes entry = entry(offset); + MultiStoreBytes entry = entry(offset); if (!keyEquals(keyWriter, metaKeyWriter, key, keySize, entry)) continue; // key is found @@ -1109,19 +1109,21 @@ class ReplicatedChronicleMap, @Override public Entry getEntry(long pos) { - bytes.position(offsetFromPos(pos) + metaDataBytes); + MultiStoreBytes entry = acquireTmpBytes(); + long offset = offsetFromPos(pos) + metaDataBytes; + entry.setBytesOffset(bytes, offset); - long keySize = keySizeMarshaller.readSize(bytes); + long keySize = keySizeMarshaller.readSize(entry); ThreadLocalCopies copies = keyReaderProvider.getCopies(null); - K key = keyReaderProvider.get(copies, originalKeyReader).read(bytes, keySize); + K key = keyReaderProvider.get(copies, originalKeyReader).read(entry, keySize); - long timestamp = bytes.readLong(); - bytes.skip(2L); // identifier and isDeleted flag + long timestamp = entry.readLong(); + entry.skip(2L); // identifier and isDeleted flag - long valueSize = valueSizeMarshaller.readSize(bytes); - alignment.alignPositionAddr(bytes); + long valueSize = valueSizeMarshaller.readSize(entry); + alignment.alignPositionAddr(entry); copies = valueReaderProvider.getCopies(copies); - V value = valueReaderProvider.get(copies, originalValueReader).read(bytes, valueSize); + V value = valueReaderProvider.get(copies, originalValueReader).read(entry, valueSize); return new TimestampTrackingEntry(key, value, timestamp); } diff --git a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java index 2b853fea3..63409557b 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java @@ -790,7 +790,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { private void writeSizeAt(long locationOfSize) { long size = bytes.position() - locationOfSize; - LOG.info("writing size = " + size); +// LOG.info("writing size = " + size); bytes.writeInt(locationOfSize, (int) size - SIZE_OF_SIZE); } diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index 35dcc94ac..786d971e8 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -1270,14 +1270,14 @@ class StatelessServerConnector { @NotNull final ByteBufferBytes reader) { - LOG.info("processStatelessEvent - writer position=" + reader.position() + +/* LOG.info("processStatelessEvent - writer position=" + reader.position() + ", " + "limit=" + reader.limit() + ",capacity=" + reader.capacity()); LOG.info("StatelessServerConnector- reader position=" + writer.position() + ", " + - "limit=" + writer.limit() + ",capacity=" + writer.capacity()); + "limit=" + writer.limit() + ",capacity=" + writer.capacity());*/ final StatelessChronicleMap.EventId event = StatelessChronicleMap.EventId.values()[eventId]; diff --git a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java index 882ddb49d..b0b386d6f 100644 --- a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java @@ -523,6 +523,14 @@ class VanillaChronicleMap, return valueSize; } + static final ThreadLocal tmpBytes = new ThreadLocal<>(); + protected MultiStoreBytes acquireTmpBytes() { + MultiStoreBytes b = tmpBytes.get(); + if (b == null) + tmpBytes.set(b = new MultiStoreBytes()); + return b; + } + // these methods should be package local, not public or private. class Segment implements SharedSegment { /* @@ -537,7 +545,6 @@ class VanillaChronicleMap, static final long SIZE_OFFSET = LOCK_OFFSET + 8L; // 32-bit final NativeBytes bytes; - final MultiStoreBytes tmpBytes = new MultiStoreBytes(); final long entriesOffset; private final int index; private final SingleThreadedDirectBitSet freeList; @@ -555,7 +562,7 @@ class VanillaChronicleMap, long start = bytes.startAddr() + SEGMENT_HEADER; createHashLookups(start); start += align64(sizeOfMultiMap() + sizeOfMultiMapBitSet()) * multiMapsPerSegment(); - final NativeBytes bsBytes = new NativeBytes(tmpBytes.objectSerializer(), + final NativeBytes bsBytes = new NativeBytes(ms.objectSerializer(), start, start + align8(align8(entriesPerSegment) / 8L), null); freeList = new SingleThreadedDirectBitSet(bsBytes); start += numberOfBitSets() * sizeOfBitSets(); @@ -624,7 +631,7 @@ class VanillaChronicleMap, public void readLock() throws IllegalStateException { while (true) { - final boolean success = bytes.tryLockNanosLong(LOCK_OFFSET, lockTimeOutNS); + final boolean success = bytes.tryRWReadLock(LOCK_OFFSET, lockTimeOutNS); if (success) return; if (currentThread().isInterrupted()) { throw new IllegalStateException(new InterruptedException("Unable to obtain lock, interrupted")); @@ -637,7 +644,7 @@ class VanillaChronicleMap, public void writeLock() throws IllegalStateException { while (true) { - final boolean success = bytes.tryLockNanosLong(LOCK_OFFSET, lockTimeOutNS); + final boolean success = bytes.tryRWWriteLock(LOCK_OFFSET, lockTimeOutNS); if (success) return; if (currentThread().isInterrupted()) { throw new IllegalStateException(new InterruptedException("Unable to obtain lock, interrupted")); @@ -650,7 +657,7 @@ class VanillaChronicleMap, public void readUnlock() { try { - bytes.unlockLong(LOCK_OFFSET); + bytes.unlockRWReadLock(LOCK_OFFSET); } catch (IllegalMonitorStateException e) { errorListener.errorOnUnlock(e); } @@ -658,7 +665,7 @@ class VanillaChronicleMap, public void writeUnlock() { try { - bytes.unlockLong(LOCK_OFFSET); + bytes.unlockRWWriteLock(LOCK_OFFSET); } catch (IllegalMonitorStateException e) { errorListener.errorOnUnlock(e); } @@ -675,15 +682,12 @@ class VanillaChronicleMap, public MultiStoreBytes entry(long offset) { - return reuse(tmpBytes, offset); + return reuse(acquireTmpBytes(), offset); } private MultiStoreBytes reuse(MultiStoreBytes entry, long offset) { offset += metaDataBytes; - entry.storePositionAndSize(bytes, offset, - // "Infinity". Limit not used when treating entries as - // possibly oversized - bytes.limit() - offset); + entry.setBytesOffset(bytes, offset); return entry; } @@ -728,7 +732,7 @@ class VanillaChronicleMap, readLock(); try { long keySize = metaKeyInterop.size(keyInterop, key); - MultiStoreBytes entry = tmpBytes; + MultiStoreBytes entry = acquireTmpBytes(); long offset = searchKey(keyInterop, metaKeyInterop, key, keySize, hash2, entry, hashLookup); if (offset >= 0L) { @@ -776,7 +780,7 @@ class VanillaChronicleMap, } V onKeyPresentOnAcquire(ThreadLocalCopies copies, K key, V usingValue, long offset, - NativeBytes entry) { + MultiStoreBytes entry) { V v = readValue(copies, entry, usingValue); notifyGet(offset, key, v); return v; @@ -817,7 +821,7 @@ class VanillaChronicleMap, hashLookup.startSearch(hash2); for (long pos; (pos = hashLookup.nextPos()) >= 0L; ) { long offset = offsetFromPos(pos); - NativeBytes entry = entry(offset); + MultiStoreBytes entry = entry(offset); if (!keyEquals(keyInterop, metaKeyInterop, key, keySize, entry)) continue; // key is found @@ -840,7 +844,7 @@ class VanillaChronicleMap, } } - V replaceValueOnPut(ThreadLocalCopies copies, K key, V value, NativeBytes entry, long pos, + V replaceValueOnPut(ThreadLocalCopies copies, K key, V value, MultiStoreBytes entry, long pos, long offset, boolean readPrevValue, MultiMap searchedHashLookup) { long valueSizePos = entry.position(); long valueSize = readValueSize(entry); @@ -978,15 +982,14 @@ class VanillaChronicleMap, nextPosToSearchFrom = fromPos; } - V readValue(ThreadLocalCopies copies, NativeBytes entry, V value) { + V readValue(ThreadLocalCopies copies, MultiStoreBytes entry, V value) { return readValue(copies, entry, value, readValueSize(entry)); } - V readValue(ThreadLocalCopies copies, NativeBytes entry, V value, long valueSize) { + V readValue(ThreadLocalCopies copies, MultiStoreBytes entry, V value, long valueSize) { copies = valueReaderProvider.getCopies(copies); BytesReader valueReader = valueReaderProvider.get(copies, originalValueReader); - bytes.positionAddr(entry.positionAddr()); - return valueReader.read(bytes, valueSize, value); + return valueReader.read(entry, valueSize, value); } boolean keyEquals(KI keyInterop, MKI metaKeyInterop, K key, @@ -1013,7 +1016,7 @@ class VanillaChronicleMap, hashLookup.startSearch(hash2); for (long pos; (pos = hashLookup.nextPos()) >= 0L; ) { long offset = offsetFromPos(pos); - NativeBytes entry = entry(offset); + MultiStoreBytes entry = entry(offset); if (!keyEquals(keyInterop, metaKeyInterop, key, keySize, entry)) continue; // key is found @@ -1074,7 +1077,7 @@ class VanillaChronicleMap, hashLookup.startSearch(hash2); for (long pos; (pos = hashLookup.nextPos()) >= 0L; ) { long offset = offsetFromPos(pos); - NativeBytes entry = entry(offset); + MultiStoreBytes entry = entry(offset); if (!keyEquals(keyInterop, metaKeyInterop, key, keySize, entry)) continue; // key is found @@ -1090,7 +1093,7 @@ class VanillaChronicleMap, } V onKeyPresentOnReplace(ThreadLocalCopies copies, K key, V expectedValue, V newValue, - long pos, long offset, NativeBytes entry, + long pos, long offset, MultiStoreBytes entry, MultiMap searchedHashLookup) { long valueSizePos = entry.position(); long valueSize = readValueSize(entry); @@ -1111,24 +1114,27 @@ class VanillaChronicleMap, void notifyPut(long offset, boolean added, K key, V value, final long pos) { if (eventListener() != MapEventListeners.NOP) { - tmpBytes.storePositionAndSize(bytes, offset, entrySize); - eventListener().onPut(VanillaChronicleMap.this, tmpBytes, metaDataBytes, + MultiStoreBytes b = acquireTmpBytes(); + b.storePositionAndSize(bytes, offset, entrySize); + eventListener().onPut(VanillaChronicleMap.this, b, metaDataBytes, added, key, value, pos, this); } } void notifyGet(long offset, K key, V value) { if (eventListener() != MapEventListeners.NOP) { - tmpBytes.storePositionAndSize(bytes, offset, entrySize); - eventListener().onGetFound(VanillaChronicleMap.this, tmpBytes, metaDataBytes, + MultiStoreBytes b = acquireTmpBytes(); + b.storePositionAndSize(bytes, offset, entrySize); + eventListener().onGetFound(VanillaChronicleMap.this, b, metaDataBytes, key, value); } } void notifyRemoved(long offset, K key, V value, final long pos) { if (eventListener() != MapEventListeners.NOP) { - tmpBytes.storePositionAndSize(bytes, offset, entrySize); - eventListener().onRemove(VanillaChronicleMap.this, tmpBytes, metaDataBytes, + MultiStoreBytes b = acquireTmpBytes(); + b.storePositionAndSize(bytes, offset, entrySize); + eventListener().onRemove(VanillaChronicleMap.this, b, metaDataBytes, key, value, pos, this); } } diff --git a/src/test/java/net/openhft/chronicle/map/ByteableIntValueMarshaller.java b/src/test/java/net/openhft/chronicle/map/ByteableIntValueMarshaller.java index f7d2d05ad..0e916b87c 100644 --- a/src/test/java/net/openhft/chronicle/map/ByteableIntValueMarshaller.java +++ b/src/test/java/net/openhft/chronicle/map/ByteableIntValueMarshaller.java @@ -18,6 +18,7 @@ package net.openhft.chronicle.map; +import net.openhft.chronicle.common.serialization.impl.ByteableMarshaller; import net.openhft.lang.io.Bytes; import net.openhft.lang.io.NativeBytes; import net.openhft.lang.io.serialization.BytesMarshaller; @@ -51,7 +52,7 @@ enum ByteableIntValueMarshaller implements BytesMarshaller { if (intValue == null) intValue = (IntValue) NativeBytes.UNSAFE.allocateInstance(DIRECT); Byteable biv = (Byteable) intValue; - biv.bytes(bytes, bytes.position()); + ByteableMarshaller.setBytesAndOffset(biv, bytes); bytes.skip(biv.maxSize()); return intValue; } catch (Exception e) { diff --git a/src/test/java/net/openhft/chronicle/map/ByteableLongValueMarshaller.java b/src/test/java/net/openhft/chronicle/map/ByteableLongValueMarshaller.java index 7190d4fd9..45e0986f3 100644 --- a/src/test/java/net/openhft/chronicle/map/ByteableLongValueMarshaller.java +++ b/src/test/java/net/openhft/chronicle/map/ByteableLongValueMarshaller.java @@ -18,6 +18,7 @@ package net.openhft.chronicle.map; +import net.openhft.chronicle.common.serialization.impl.ByteableMarshaller; import net.openhft.lang.io.Bytes; import net.openhft.lang.io.NativeBytes; import net.openhft.lang.io.serialization.BytesMarshaller; @@ -51,7 +52,7 @@ enum ByteableLongValueMarshaller implements BytesMarshaller { if (longValue == null) longValue = (LongValue) NativeBytes.UNSAFE.allocateInstance(DIRECT); Byteable biv = (Byteable) longValue; - biv.bytes(bytes, bytes.position()); + ByteableMarshaller.setBytesAndOffset(biv, bytes); bytes.skip(biv.maxSize()); return longValue; } catch (Exception e) { diff --git a/src/test/java/net/openhft/chronicle/map/ChronicleMapTest.java b/src/test/java/net/openhft/chronicle/map/ChronicleMapTest.java index d0bc73631..5dfd1d01f 100644 --- a/src/test/java/net/openhft/chronicle/map/ChronicleMapTest.java +++ b/src/test/java/net/openhft/chronicle/map/ChronicleMapTest.java @@ -470,8 +470,8 @@ public class ChronicleMapTest { @Test public void testGetWithoutAcquireFirst() throws Exception { - ChronicleMap map = getSharedMap(10 * 1000, 128, 24); - assertNull(map.getUsing("key", new LongValue$$Native())); + ChronicleMap map = getSharedMap(10 , 1, 24); + assertNull(map.getUsing("key", DataValueClasses.newDirectReference(LongValue.class))); map.close(); } @@ -491,10 +491,10 @@ public class ChronicleMapTest { @Test public void testAcquireAndGet() throws IOException, ClassNotFoundException, IllegalAccessException, InstantiationException { - int entries = 100 * 1000; - testAcquireAndGet(getSharedMap(entries, 128, 24, NO_ALIGNMENT), entries); - testAcquireAndGet(getSharedMap(entries, 128, 24, OF_4_BYTES), entries); - testAcquireAndGet(getSharedMap(entries, 128, 24, OF_8_BYTES), entries); + int entries = 3/*00 * 1000*/; + testAcquireAndGet(getSharedMap(entries, 1, 24, OF_4_BYTES), entries); + testAcquireAndGet(getSharedMap(entries, 1, 24, NO_ALIGNMENT), entries); + testAcquireAndGet(getSharedMap(entries, 1, 24, OF_8_BYTES), entries); } public void testAcquireAndGet(ChronicleMap map, int entries) diff --git a/src/test/java/net/openhft/chronicle/map/IntIntMultiMapTest.java b/src/test/java/net/openhft/chronicle/map/IntIntMultiMapTest.java index e43f8a312..18d5c8686 100644 --- a/src/test/java/net/openhft/chronicle/map/IntIntMultiMapTest.java +++ b/src/test/java/net/openhft/chronicle/map/IntIntMultiMapTest.java @@ -23,10 +23,13 @@ import com.google.common.collect.HashMultimap; import com.google.common.collect.Multimap; import net.openhft.lang.collection.SingleThreadedDirectBitSet; import net.openhft.lang.io.DirectStore; +import org.junit.Ignore; import org.junit.Test; import org.junit.runner.RunWith; import org.junit.runners.Parameterized; +import java.io.File; +import java.io.FileNotFoundException; import java.lang.reflect.Field; import java.util.*; @@ -51,8 +54,8 @@ public class IntIntMultiMapTest { @Parameterized.Parameters public static Collection data() { return Arrays.asList(new Object[][]{ + {VanillaShortShortMultiMap.class}, {VanillaIntIntMultiMap.class}, - {VanillaShortShortMultiMap.class} }); } @@ -150,13 +153,14 @@ public class IntIntMultiMapTest { } @Test + @Ignore("Very long running test") public void testMaxCapacity() throws NoSuchFieldException, IllegalAccessException { Field maxCapacityField = c.getDeclaredField("MAX_CAPACITY"); long maxCapacity = maxCapacityField.getLong(null); Field entrySizeField = c.getDeclaredField("ENTRY_SIZE"); entrySizeField.setAccessible(true); long entrySize = entrySizeField.getLong(null); - if (Runtime.getRuntime().maxMemory() < maxCapacity * entrySize * 3L / 2L) { + if (maxMemory() * 1024L < maxCapacity * entrySize * 3L / 2L) { System.out.println("Skipped " + c + " maxCapacity test because there is not enough " + "memory in system"); return; @@ -177,4 +181,16 @@ public class IntIntMultiMapTest { }); assertTrue(bs.allSet(0L, maxCapacity)); } + + private long maxMemory() { + File meminfo = new File("/proc/meminfo"); + try { + try (Scanner sc = new Scanner(meminfo)) { + sc.next(); + return sc.nextLong(); + } + } catch (FileNotFoundException e) { + return -1; + } + } } diff --git a/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationIntValueTest.java b/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationIntValueTest.java index 3c34e0d2a..982d5ca78 100644 --- a/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationIntValueTest.java +++ b/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationIntValueTest.java @@ -19,8 +19,9 @@ package net.openhft.chronicle.map; import net.openhft.lang.io.ByteBufferBytes; +import net.openhft.lang.model.Byteable; +import net.openhft.lang.model.DataValueClasses; import net.openhft.lang.values.IntValue; -import net.openhft.lang.values.IntValue$$Native; import org.junit.After; import org.junit.Before; import org.junit.Test; @@ -45,16 +46,15 @@ import static org.junit.Assert.assertTrue; public class TCPSocketReplicationIntValueTest { - private ChronicleMapBuilder map1Builder; private ChronicleMap map1; private ChronicleMap map2; - private IntValue$$Native value; + private IntValue value; @Before public void setup() throws IOException { - value = new IntValue$$Native(); - value.bytes(new ByteBufferBytes(ByteBuffer.allocateDirect(4)), 0); + value = DataValueClasses.newDirectReference(IntValue.class); + ((Byteable) value).bytes(new ByteBufferBytes(ByteBuffer.allocateDirect(4)), 0); map1Builder = newTcpSocketShmBuilder(IntValue.class, CharSequence.class, (byte) 1, 8076, new InetSocketAddress("localhost", 8077)); map1Builder.keyMarshaller(ByteableIntValueMarshaller.INSTANCE).file(getPersistenceFile()); @@ -88,7 +88,7 @@ public class TCPSocketReplicationIntValueTest { assertTrue(!map1.isEmpty()); } - private IntValue$$Native set(int x) { + private IntValue set(int x) { value.setValue(x); return value; diff --git a/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationTest3VoidValueTest.java b/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationTest3VoidValueTest.java index 10311205b..8f532bc9f 100644 --- a/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationTest3VoidValueTest.java +++ b/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationTest3VoidValueTest.java @@ -112,7 +112,7 @@ public class TCPSocketReplicationTest3VoidValueTest { map2.put(5, null); // allow time for the recompilation to resolve - assertTrue(waitTillEqual(5000)); + assertTrue("test timed out", waitTillEqual(5000)); assertEquals(map3, map3); assertTrue(!map1.isEmpty());