Alawn/Chronicle-Map
1
0
Fork 0
mirror of https://github.com/OpenHFT/Chronicle-Map.git synced 2021-09-13 01:26:12 +00:00
Code Releases Activity

Remote udpates remove positions from modifications iterators (previously was not touching modification iterators; fixed a small bug in ReplicatedCHronicleMap.removeWithoutUnlock(); fixed soak tests with random data

Browse Source
This commit is contained in:
leventov
2014-12-03 20:26:52 +07:00
parent 2e2798f318
commit ec6e99224d
5 changed files with 53 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java
+13 -3
View File
@@ -387,6 +387,11 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
}
}
@Override
void onRemotePut(VanillaChronicleMap<K, KI, MKI, V, VI, MVI>.Segment segment, long pos) {
onRelocation(segment, pos); // erase modification bits
}
@Override
void onRemove(VanillaChronicleMap<K, KI, MKI, V, VI, MVI>.Segment segment, long pos) {
for (long next = modIterSet.nextSetBit(0L); next > 0L;
@@ -399,6 +404,11 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
}
}
@Override
void onRemoteRemove(VanillaChronicleMap<K, KI, MKI, V, VI, MVI>.Segment segment, long pos) {
onRelocation(segment, pos); // erase modification bits
}
@Override
void onRelocation(VanillaChronicleMap<K, KI, MKI, V, VI, MVI>.Segment segment, long pos) {
for (long next = modIterSet.nextSetBit(0L); next > 0L;
@@ -931,8 +941,7 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
entry.writeBoolean(false);
// put callbacks
if (!remote)
onPut(this, localSegmentState.pos);
onPutMaybeRemote(localSegmentState.pos, remote);
if (bytesEventListener != null)
bytesEventListener.onPut(entry, 0L, metaDataBytes,
localSegmentState.valueSizePos, true);
@@ -1086,6 +1095,7 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
entry.position(timestampPos);
entry.writeLong(timestamp);
entry.writeByte(identifier);
onRemoveMaybeRemote(pos, remote);
break; // to "key not found" location
}
@@ -1452,7 +1462,7 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
}
// if the position was already cleared by another thread
// while we were trying to obtain segment lock (for example, in onReplication()),
// while we were trying to obtain segment lock (for example, in onRelocation()),
// go to pick up next (next iteration in while (true) loop)
} finally {
segment.readUnlock();
src/main/java/net/openhft/chronicle/map/TcpReplicator.java
+3
View File
@@ -203,6 +203,9 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
LOG.debug("", e);
} catch (Exception e) {
LOG.error("", e);
} catch (Throwable e) {
LOG.error("", e);
throw e;
} finally {
if (LOG.isDebugEnabled())
src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java
+30 -8
View File
@@ -1368,10 +1368,18 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
// do nothing
}
void onRemoteRemove(Segment segment, long pos) {
// do nothing
}
void onPut(Segment segment, long pos) {
// do nothing
}
void onRemotePut(Segment segment, long pos) {
// do nothing
}
void onRelocation(Segment segment, long pos) {
// do nothing
}
@@ -1940,8 +1948,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
metaValueInterop, valueInterop, value, valueSize, searchedHashLookup);
// put callbacks
if (!remote)
onPut(this, segmentState.pos);
onPutMaybeRemote(segmentState.pos, remote);
if (bytesEventListener != null)
bytesEventListener.onPut(entry, 0L, metaDataBytes, valueSizePos, false);
if (eventListener != null) {
@@ -1952,6 +1959,14 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
return resultUnused ? null : prevValue;
}
final void onPutMaybeRemote(long pos, boolean remote) {
if (remote) {
onRemotePut(this, pos);
} else {
onPut(this, pos);
}
}
/**
* Returns value size, writes the entry (key, value, sizes) to the entry, after this method
* call entry positioned after value bytes written (i. e. at the end of entry), sets entry
@@ -2169,8 +2184,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
decrementSize();
// remove callbacks
if (!remote)
onRemove(this, pos);
onRemoveMaybeRemote(pos, remote);
if (bytesEventListener != null)
bytesEventListener.onRemove(entry, 0L, metaDataBytes, valueSizePos);
if (eventListener != null) {
@@ -2183,6 +2197,14 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
return booleanResult ? Boolean.TRUE : removedValue;
}
final void onRemoveMaybeRemote(long pos, boolean remote) {
if (remote) {
onRemoteRemove(this, pos);
} else {
onRemove(this, pos);
}
}
private <KB, KBI, MKBI extends MetaBytesInterop<KB, ? super KBI>>
boolean containsKey(ThreadLocalCopies copies,
MKBI metaKeyInterop, KBI keyInterop, KB key, long keySize, long hash2) {
@@ -2337,15 +2359,15 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
MetaBytesWriter<E, ? super EW> metaElemWriter, EW elemWriter, E newElem,
long newElemSize,
MultiMap searchedHashLookup) {
long valueSizeAddr = entry.address() + valueSizePos;
long entryStartAddr = entry.address();
long valueSizeAddr = entryStartAddr + valueSizePos;
long newValueAddr = alignment.alignAddr(
valueSizeAddr + valueSizeMarshaller.sizeEncodingSize(newElemSize));
long newEntryEndAddr = newValueAddr + newElemSize;
long entryStartAddr = entry.address();
long oldEntrySize = entryEndAddr - entryStartAddr;
int oldSizeInBlocks = inBlocks(oldEntrySize);
newValueDoesNotFit:
if (newEntryEndAddr != entryEndAddr) {
long oldEntrySize = entryEndAddr - entryStartAddr;
int oldSizeInBlocks = inBlocks(oldEntrySize);
int newSizeInBlocks = inBlocks(newEntryEndAddr - entryStartAddr);
if (newSizeInBlocks > oldSizeInBlocks) {
if (newSizeInBlocks > MAX_ENTRY_OVERSIZE_FACTOR) {
src/test/java/net/openhft/chronicle/map/TCPSocketReplicationIntValueTest.java
+1 -2
View File
@@ -161,14 +161,13 @@ public class TCPSocketReplicationIntValueTest {
for (; t < timeOutMs; t++) {
if (map1.equals(map2))
break;
Thread.sleep(25);
Thread.sleep(1);
}
assertEquals(map1, map2);
}
// TODO test this with larger sizes.
@Test(timeout = 12000)
@Ignore("Doesn't work, maps need to check equality.")
public void testSoakTestWithRandomData() throws IOException, InterruptedException {
final long start = System.currentTimeMillis();
System.out.print("SoakTesting ");
src/test/java/net/openhft/chronicle/map/TcpReplicationSoakTest.java
+6 -4
View File
@@ -12,6 +12,7 @@ import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.util.*;
import java.util.concurrent.TimeUnit;
/**
* @author Rob Austin.
@@ -33,7 +34,8 @@ public class TcpReplicationSoakTest {
{
final TcpTransportAndNetworkConfig tcpConfig1 = TcpTransportAndNetworkConfig.of(s_port,
endpoint).autoReconnectedUponDroppedConnection(true).name(" map1");
endpoint).autoReconnectedUponDroppedConnection(true).name(" map1")
.heartBeatInterval(1, TimeUnit.SECONDS);
map1 = ChronicleMapBuilder.of(Integer.class, CharSequence.class)
@@ -45,7 +47,8 @@ public class TcpReplicationSoakTest {
}
{
final TcpTransportAndNetworkConfig tcpConfig2 = TcpTransportAndNetworkConfig.of
(s_port + 1).autoReconnectedUponDroppedConnection(true).name("map2");
(s_port + 1).autoReconnectedUponDroppedConnection(true).name("map2")
.heartBeatInterval(1, TimeUnit.SECONDS);
map2 = ChronicleMapBuilder.of(Integer.class, CharSequence.class)
.entries(Builder.SIZE + Builder.SIZE)
@@ -86,7 +89,6 @@ public class TcpReplicationSoakTest {
@Test
@Ignore
public void testSoakTestWithRandomData() throws IOException, InterruptedException {
System.out.print("SoakTesting ");
@@ -130,7 +132,7 @@ public class TcpReplicationSoakTest {
map1UnChanged = new HashMap(map1);
map2UnChanged = new HashMap(map2);
}
Thread.sleep(500);
Thread.sleep(1);
if (numberOfTimesTheSame == 100) {
System.out.println("same");
break;