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

Merge pull request #225 from Cotton-Ben/ACID=transactions=Isolation=Tests

Browse Source 
Acid=transactions=isolation=tests
This commit is contained in:
Rob Austin
2020-07-14 17:12:26 +01:00
committed by GitHub
parent 67b612c256 065b549b9f
commit dcf7446293
36 changed files with 1961 additions and 291 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+1
View File
@@ -52,3 +52,4 @@ local.properties
### Queue files
*.cq4t
*.cq4
JCACHE.pdf
BIN
View File
Binary file not shown.
Santos.pdf
BIN
View File
Binary file not shown.
basEC.png
BIN
View File
Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 60 KiB

capTradeoffs=Chronicle=BasEC.png
BIN
View File
Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 104 KiB

cp.txt
+1
View File
File diff suppressed because one or more lines are too long
cp2.txt
+1
View File
File diff suppressed because one or more lines are too long
dr.csv
+17
View File
@@ -0,0 +1,17 @@
DirtyReadOffender,chrAig[DIRTY_READ_OPTIMISTIC]coupon=3.50,DirtyReadVictim
,Established chm,
,,@t=1593597010401 DirtyReadVictim CALLING offHeapLock.tryOptimisticRead()
,,@t=1593597010402 DirtyReadVictim CALLED offHeapLock.tryOptimisticRead()
,,@t=1593597010803 DirtyReadVictim calling chm.get('369604101').getCoupon()
,,@t=1593597010805 DirtyReadVictim coupon=[4.096316014939393] read.
,,@t=1593597010805 DirtyReadVictim sleeping 60 seconds
@t=1593597010821 DirtyReadOffender sleeping 10 seconds
@t=1593597020822 DirtyReadOffender awakening
@t=1593597020822 DirtyReadOffender ACQUIRING offHeapLock.writeLock();
@t=1593597020822 DirtyReadOffender ACQUIRED offHeapLock.writeLock();
@t=1593597020823 DirtyReadOffender calling chm.put('369604101',3.5085267913490994)
@t=1593597020827 DirtyReadOffender coupon=[3.5085267913490994] written.
@t=1593597020828 DirtyReadOffender called offHeapLock.unlockWrite(stamp);
,,@t=1593597070806 DirtyReadVictim FAILED offHeapLock.validate(stamp) must apply PESSIMISTIC_POLICY (dirty read endured) coupon=[4.096316014939393] is *DIRTY*.
,,@t=1593597070806 DirtyReadVictim got() coupon=4.096316014939393
,,@t=1593597070806 DirtyReadVictim COMMITTED
1 DirtyReadOffender,chrAig[DIRTY_READ_OPTIMISTIC]coupon=3.50,DirtyReadVictim
2 ,Established chm,
3 ,,@t=1593597010401 DirtyReadVictim CALLING offHeapLock.tryOptimisticRead()
4 ,,@t=1593597010402 DirtyReadVictim CALLED offHeapLock.tryOptimisticRead()
5 ,,@t=1593597010803 DirtyReadVictim calling chm.get('369604101').getCoupon()
6 ,,@t=1593597010805 DirtyReadVictim coupon=[4.096316014939393] read.
7 ,,@t=1593597010805 DirtyReadVictim sleeping 60 seconds
8 @t=1593597010821 DirtyReadOffender sleeping 10 seconds
9 @t=1593597020822 DirtyReadOffender awakening
10 @t=1593597020822 DirtyReadOffender ACQUIRING offHeapLock.writeLock();
11 @t=1593597020822 DirtyReadOffender ACQUIRED offHeapLock.writeLock();
12 @t=1593597020823 DirtyReadOffender calling chm.put('369604101',3.5085267913490994)
13 @t=1593597020827 DirtyReadOffender coupon=[3.5085267913490994] written.
14 @t=1593597020828 DirtyReadOffender called offHeapLock.unlockWrite(stamp);
15 ,,@t=1593597070806 DirtyReadVictim FAILED offHeapLock.validate(stamp) must apply PESSIMISTIC_POLICY (dirty read endured) coupon=[4.096316014939393] is *DIRTY*.
16 ,,@t=1593597070806 DirtyReadVictim got() coupon=4.096316014939393
17 ,,@t=1593597070806 DirtyReadVictim COMMITTED
intolerant.csv
+23
View File
@@ -0,0 +1,23 @@
DirtyReadOffender,chrAig[DIRTY_READ_INTOLERANT]coupon=3.50,DirtyReadVictim
,Established chrAig,
,,@t=1588854013435 DirtyReadVictim sleeping 10 seconds
@t=1588854013450 DirtyReadOffender calling chrAig.put('369604101'/3.85) ,,
, @t=1588854013450 Tx=Thread[Thread-4.5.main] inside chrAig.put('369604101'/null) BEGIN,
, @t=1588854013450 Tx=Thread[Thread-4.5.main] inside chrAig.put('369604101'/3.85) DONE,
@t=1588854013450 DirtyReadOffender sleeping 1 minute. ,,
,,@t=1588854023448 DirtyReadVictim calling chrAig.get('369604101')
, @t=1588854023448 Tx=Thread[Thread-5.5.main] inside chrAig.get('369604101') BEGIN,
, @t=1588854023448 Tx=Thread[Thread-5.5.main] inside chrAig.get() WAITING ,
@t=1588854073452 DirtyReadOffender calling chrAig.rollback() ,,
, @t=1588854073452 Tx=Thread[Thread-4.5.main] chrAig.rollback() BEGIN ,
,@t=1588854073452 Tx=Thread[Thread-4.5.main] chrAig.rollback() COMPLETE coupon=3.50,
, @t=1588854073452 Tx=Thread[Thread-4.5.main] chrAig.rollback() completed notifyAll() to waiting Tx Threads,
, @t=1588854073452 Tx=Thread[Thread-5.5.main] inside chrAig.get() RESUMING,
, @t=1588854073452 Tx=Thread[Thread-5.5.main] inside chrAig.get() DONE ,
,,@t=1588854073452 DirtyReadVictim got() coupon=3.5
@t=1588854073452 DirtyReadOffender ROLLED BACK ,,
,,@t=1588854073452 DirtyReadVictim calling chrAig.commmit()
,@t=1588854073452 Tx=Thread[Thread-5.5.main] chrAig.commit() BEGIN ,
, @t=1588854073452 Tx=Thread[Thread-5.5.main] chrAig.commit() END ,
, @t=1588854073452 Tx=Thread[Thread-5.5.main] chrAig.commit() complete notifyAll() to waiting Tx Threads ,
,,@t=1588854073452 DirtyReadVictim COMMITTED
1 DirtyReadOffender chrAig[DIRTY_READ_INTOLERANT]coupon=3.50 DirtyReadVictim
2 Established chrAig
3 @t=1588854013435 DirtyReadVictim sleeping 10 seconds
4 @t=1588854013450 DirtyReadOffender calling chrAig.put('369604101'/3.85)
5 @t=1588854013450 Tx=Thread[Thread-4.5.main] inside chrAig.put('369604101'/null) BEGIN
6 @t=1588854013450 Tx=Thread[Thread-4.5.main] inside chrAig.put('369604101'/3.85) DONE
7 @t=1588854013450 DirtyReadOffender sleeping 1 minute.
8 @t=1588854023448 DirtyReadVictim calling chrAig.get('369604101')
9 @t=1588854023448 Tx=Thread[Thread-5.5.main] inside chrAig.get('369604101') BEGIN
10 @t=1588854023448 Tx=Thread[Thread-5.5.main] inside chrAig.get() WAITING
11 @t=1588854073452 DirtyReadOffender calling chrAig.rollback()
12 @t=1588854073452 Tx=Thread[Thread-4.5.main] chrAig.rollback() BEGIN
13 @t=1588854073452 Tx=Thread[Thread-4.5.main] chrAig.rollback() COMPLETE coupon=3.50
14 @t=1588854073452 Tx=Thread[Thread-4.5.main] chrAig.rollback() completed notifyAll() to waiting Tx Threads
15 @t=1588854073452 Tx=Thread[Thread-5.5.main] inside chrAig.get() RESUMING
16 @t=1588854073452 Tx=Thread[Thread-5.5.main] inside chrAig.get() DONE
17 @t=1588854073452 DirtyReadVictim got() coupon=3.5
18 @t=1588854073452 DirtyReadOffender ROLLED BACK
19 @t=1588854073452 DirtyReadVictim calling chrAig.commmit()
20 @t=1588854073452 Tx=Thread[Thread-5.5.main] chrAig.commit() BEGIN
21 @t=1588854073452 Tx=Thread[Thread-5.5.main] chrAig.commit() END
22 @t=1588854073452 Tx=Thread[Thread-5.5.main] chrAig.commit() complete notifyAll() to waiting Tx Threads
23 @t=1588854073452 DirtyReadVictim COMMITTED
optimistic.csv
+16
View File
@@ -0,0 +1,16 @@
DirtyReadOffender,chrAig[DIRTY_READ_OPTIMISTIC]coupon=3.50,DirtyReadVictim
,Established chrAig,
,,@t=1593467845991 DirtyReadVictim sleeping 10 seconds
@t=1593467846072 DirtyReadOffender calling chrAig.put('369604101'/3.85)
, @t=1593467846079 Tx=Thread[Thread-2.5.main] inside chrAig.put('369604101'/null) BEGIN,
, @t=1593467846083 Tx=Thread[Thread-2.5.main] inside chrAig.put('369604101'/3.85) DONE,
@t=1593467846083 DirtyReadOffender sleeping 1 minute.
,,@t=1593467856000 DirtyReadVictim calling chrAig.get('369604101')
, @t=1593467856002 Tx=Thread[Thread-3.5.main] inside chrAig.get('369604101') BEGIN,
, @t=1593467856036 Tx=Thread[Thread-3.5.main] inside chrAig.get() DONE ,
,,@t=1593467856036 DirtyReadVictim OPTIMISTICALLY_READ coupon=3.85
,,@t=1593467856036 DirtyReadVictim calling chrAig.rollback()
, @t=1593467856036 Tx=Thread[Thread-3.5.main] chrAig.rollback() BEGIN ,
,@t=1593467856037 Tx=Thread[Thread-3.5.main] chrAig.rollback() COMPLETE coupon=3.50,
, @t=1593467856037 Tx=Thread[Thread-3.5.main] chrAig.rollback() completed notifyAll() to waiting Tx Threads,
,,@t=1593467856037 DirtyReadVictim ROLLED BACK
1 DirtyReadOffender,chrAig[DIRTY_READ_OPTIMISTIC]coupon=3.50,DirtyReadVictim
2 ,Established chrAig,
3 ,,@t=1593467845991 DirtyReadVictim sleeping 10 seconds
4 @t=1593467846072 DirtyReadOffender calling chrAig.put('369604101'/3.85)
5 , @t=1593467846079 Tx=Thread[Thread-2.5.main] inside chrAig.put('369604101'/null) BEGIN,
6 , @t=1593467846083 Tx=Thread[Thread-2.5.main] inside chrAig.put('369604101'/3.85) DONE,
7 @t=1593467846083 DirtyReadOffender sleeping 1 minute.
8 ,,@t=1593467856000 DirtyReadVictim calling chrAig.get('369604101')
9 , @t=1593467856002 Tx=Thread[Thread-3.5.main] inside chrAig.get('369604101') BEGIN,
10 , @t=1593467856036 Tx=Thread[Thread-3.5.main] inside chrAig.get() DONE ,
11 ,,@t=1593467856036 DirtyReadVictim OPTIMISTICALLY_READ coupon=3.85
12 ,,@t=1593467856036 DirtyReadVictim calling chrAig.rollback()
13 , @t=1593467856036 Tx=Thread[Thread-3.5.main] chrAig.rollback() BEGIN ,
14 ,@t=1593467856037 Tx=Thread[Thread-3.5.main] chrAig.rollback() COMPLETE coupon=3.50,
15 , @t=1593467856037 Tx=Thread[Thread-3.5.main] chrAig.rollback() completed notifyAll() to waiting Tx Threads,
16 ,,@t=1593467856037 DirtyReadVictim ROLLED BACK
pom.xml
+20
View File
@@ -1,6 +1,7 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
@@ -11,6 +12,7 @@
</parent>
<artifactId>chronicle-map</artifactId>
<version>3.19.43-SNAPSHOT</version>
<name>OpenHFT/Chronicle-Map</name>
<description>Chronicle-Map</description>
@@ -225,6 +227,24 @@
<version>1.11</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.junit.jupiter</groupId>
<artifactId>junit-jupiter</artifactId>
<version>RELEASE</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.junit.jupiter</groupId>
<artifactId>junit-jupiter</artifactId>
<version>RELEASE</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.junit.jupiter</groupId>
<artifactId>junit-jupiter</artifactId>
<version>RELEASE</version>
<scope>test</scope>
</dependency>
</dependencies>
pwd
View File
src/test/java/net/openhft/chronicle/map/CHMTest5.java
-224
View File
@@ -1,224 +0,0 @@
/*
* Copyright 2012-2018 Chronicle Map Contributors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package net.openhft.chronicle.map;
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;
import java.io.File;
import java.io.IOException;
import java.util.concurrent.TimeUnit;
import static net.openhft.chronicle.algo.bytes.Access.checkedBytesStoreAccess;
import static org.junit.Assert.assertEquals;
/**
* @author jack shirazi
*/
public class CHMTest5 {
public static final String TEST_KEY = "whatever";
public static int NUMBER_OF_PROCESSES_ALLOWED = 2;
public static void main(String[] args) throws IOException {
//First create (or access if already created) the shared map
ChronicleMapBuilder<String, CHMTest5Data> builder =
ChronicleMapBuilder.of(String.class, CHMTest5Data.class)
.entries(1000);
//// don't include this, just to check it is as expected.
assertEquals(8, builder.minSegments());
//// end of test
String chmPath = System.getProperty("java.io.tmpdir") + System.getProperty("file.separator") + "CHMTest5";
ChronicleMap<String, CHMTest5Data> theSharedMap =
builder.createPersistedTo(new File(chmPath));
//Now get the shared data object, auto-creating if it's not there
CHMTest5Data data = Values.newNativeReference(CHMTest5Data.class);
theSharedMap.acquireUsing(TEST_KEY, data);
//if this was newly created, we need to set the max allowed
//Note, the object is pointing at shared memory and writes
//directly to it, so no need to put() it back into the map
if (data.getMaxNumberOfProcessesAllowed() != NUMBER_OF_PROCESSES_ALLOWED) {
//it's either a new object, set to 0, or
//another process set it to an invalid value
if (data.compareAndSwapMaxNumberOfProcessesAllowed(0, NUMBER_OF_PROCESSES_ALLOWED)) {
//What we expected, everything's good
} else {
//something else set a value, if it's not 2 we've got a conflict
if (data.getMaxNumberOfProcessesAllowed() != NUMBER_OF_PROCESSES_ALLOWED) {
System.out.println("Incorrect configuration found, expected " +
NUMBER_OF_PROCESSES_ALLOWED + " slots, instead found " +
data.getMaxNumberOfProcessesAllowed() + " - exiting");
System.exit(0);
}
}
}
//Now, we look for an empty slot in the "time" array.
//Because this initial implementation of the data object
//is directly accessing the shared memory, we just operate
//on it as a shared object
// There would be no need to lock the MaxNumberOfProcessesAllowed field
// as we used CAS to update and it's read-only after that
// but we need to lock access to the Time array
long[] times1 = new long[NUMBER_OF_PROCESSES_ALLOWED];
getTimes(data, times1);
pause(300L);
long[] times2 = new long[NUMBER_OF_PROCESSES_ALLOWED];
getTimes(data, times2);
//look for a slot that hasn't changed in that 300ms pause
int slotindex = 0;
long lastUpdateTime = -1;
for (; slotindex < times1.length; slotindex++) {
if (times2[slotindex] == times1[slotindex]) {
//we have an index which has not been updated by anything else
//in the 300ms pause, so we have a spare slot - we use this slot
long timenow = System.currentTimeMillis();
tryLock1Sec(data);
try {
data.setTimeAt(slotindex, timenow);
} finally {
//and release the lock
releaseLock(data);
}
//Now we have successfully acquired a slot
lastUpdateTime = timenow;
System.out.println("We have started on slot " + slotindex);
break;
}
}
if (lastUpdateTime == -1) {
System.out.println("We failed to find a free slot, so terminating now");
System.exit(0);
}
//Now let's run for 60 seconds, updating the slot every 100ms
for (int count = 0; count < 600; count++) {
pause(100L);
long timenow = System.currentTimeMillis();
tryLock1Sec(data);
try {
if (lastUpdateTime == data.getTimeAt(slotindex)) {
//That's what we expect so just update the slot
data.setTimeAt(slotindex, timenow);
lastUpdateTime = timenow;
} else {
//Some other process has hijacked our slot - so
//the only thing we can do is terminate
System.out.println("Another process has hijacked our slot " + slotindex + ", so terminating now");
System.exit(0);
}
} finally {
releaseLock(data);
}
}
System.out.println("Exiting slot " + slotindex + " after completing the full test.");
}
private static void releaseLock(CHMTest5Data data) {
//and release the lock
try {
data.unlockEntry();
} catch (IllegalMonitorStateException e) {
//odd, but we'll be unlocked either way
System.out.println("Unexpected state: " + e);
e.printStackTrace();
}
}
private static void tryLock1Sec(CHMTest5Data data) {
boolean locked = false;
for (int i = 0; i < 1000000; i++) {
//try up to 1 second
if (data.tryLockNanosEntry(1000L)) {
locked = true;
break;
}
}
if (!locked) {
System.out.println("Unable to acquire a lock on the time array - exiting");
System.exit(0);
}
}
private static void getTimes(CHMTest5Data data, long[] times1) {
tryLock1Sec(data);
try {
//we've got the lock, now copy the array
for (int i = 0; i < times1.length; i++) {
times1[i] = data.getTimeAt(i);
}
} finally {
//and release the lock
releaseLock(data);
}
}
public static void pause(long pause) {
ProcessInstanceLimiter.pause(pause);
}
public interface CHMTest5Data {
@Group(0)
long getEntryLockState();
void setEntryLockState(long entryLockState);
@Group(1)
int getMaxNumberOfProcessesAllowed();
void setMaxNumberOfProcessesAllowed(int max);
boolean compareAndSwapMaxNumberOfProcessesAllowed(int expected, int value);
@Group(1)
@Array(length = 4)
void setTimeAt(int index, long time);
long getTimeAt(int index);
@Deprecated()
default boolean tryLockNanosEntry(long nanos) {
return AcquisitionStrategies
.<ReadWriteLockingStrategy>spinLoop(nanos, TimeUnit.NANOSECONDS).acquire(
TryAcquireOperations.writeLock(),
VanillaReadWriteWithWaitsLockingStrategy.instance(),
checkedBytesStoreAccess(), ((Byteable) this).bytesStore(),
((Byteable) this).offset());
}
@Deprecated()
default void unlockEntry() throws IllegalMonitorStateException {
VanillaReadWriteWithWaitsLockingStrategy.instance()
.writeUnlock(checkedBytesStoreAccess(),
((Byteable) this).bytesStore(), ((Byteable) this).offset());
}
}
}
src/test/java/net/openhft/chronicle/map/ExitHookTest.java
+4 -2
View File
@@ -18,6 +18,7 @@ package net.openhft.chronicle.map;
import com.google.common.base.Preconditions;
import net.openhft.chronicle.core.Jvm;
import net.openhft.chronicle.core.OS;
import org.junit.Rule;
import org.junit.Test;
@@ -35,7 +36,8 @@ import static org.junit.Assert.*;
public class ExitHookTest {
private static final int KEY = 1;
private static final int JVM_STARTUP_WAIT_TIME_MS = 2_000;
private static final int JVM_STARTUP_WAIT_TIME_MS = 5_000;
private static final String PRE_SHUTDOWN_ACTION_EXECUTED = "PRE_SHUTDOWN_ACTION_EXECUTED";
private static final String USER_SHUTDOWN_HOOK_EXECUTED = "USER_SHUTDOWN_HOOK_EXECUTED";
@@ -92,7 +94,7 @@ public class ExitHookTest {
mapReference.set(mapBuilder.createPersistedTo(mapFile));
try (ExternalMapQueryContext<Integer, Integer, ?> c = mapReference.get().queryContext(KEY)) {
c.writeLock().lock();
Thread.sleep(8_000);
Thread.sleep(30_000);
}
}
src/test/java/net/openhft/chronicle/map/fromdocs/acid/ChronicleAcidIsolation.java
+13
View File
@@ -0,0 +1,13 @@
package net.openhft.chronicle.map.fromdocs.acid;
public interface ChronicleAcidIsolation extends java.sql.Connection{
int LOWEST_LATENCY = ChronicleAcidIsolation.TRANSACTION_NONE;
int DIRTY_READ_OPTIMISTIC = ChronicleAcidIsolation.TRANSACTION_READ_UNCOMMITTED;
int DIRTY_READ_INTOLERANT = ChronicleAcidIsolation.TRANSACTION_READ_COMMITTED;
int REPEATABLE_READS_MANDATORY = ChronicleAcidIsolation.TRANSACTION_REPEATABLE_READ;
int PHANTOM_READ_INTOLERANT = ChronicleAcidIsolation.TRANSACTION_SERIALIZABLE;
}
src/test/java/net/openhft/chronicle/map/fromdocs/acid/ChronicleAcidIsolationGovernor.java
+436
View File
@@ -0,0 +1,436 @@
package net.openhft.chronicle.map.fromdocs.acid;
//import net.openhft.affinity.AffinitySupport;
import net.openhft.chronicle.map.ChronicleMap;
import net.openhft.chronicle.map.fromdocs.BondVOInterface;
import java.sql.*;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.Executor;
public class ChronicleAcidIsolationGovernor implements ChronicleAcidIsolation {
private ChronicleMap<String, Integer> transactionIsolationMap;
private ChronicleMap<String, BondVOInterface> compositeChronicleMap; //hacked in, not generic
private String priorCusip;
private Double aCoupon;
public synchronized void put(String cusip, BondVOInterface bond) throws Exception {
System.out.println(
" , @t="+System.currentTimeMillis()+
" Tx="+Thread.currentThread().toString().replaceAll(",",".")+
" inside chrAig.put('"+cusip+"'/"+aCoupon+") BEGIN"+
", "
);
ChronicleMap<String, BondVOInterface> cMap = this.getCompositeChronicleMap();
this.aCoupon = cMap.get(cusip).getCoupon();
bond.setCoupon(aCoupon);
cMap.put(cusip, bond);
System.out.println(
" , @t="+System.currentTimeMillis()+
" Tx="+Thread.currentThread().toString().replaceAll(",",".")+
" inside chrAig.put('"+cusip+"'/"+aCoupon+") DONE"+
", "
);
}
//here is where the drama happens
public synchronized BondVOInterface get(String cusip) throws Exception {
System.out.println(
" , @t="+System.currentTimeMillis()+
" Tx="+Thread.currentThread().toString().replaceAll(",",".")+
" inside chrAig.get('"+cusip+"') BEGIN"+
", "
);
BondVOInterface b = null;
String tx = Thread.currentThread().toString();
ChronicleMap<String,Integer> txMap = this.getTransactionIsolationMap();
if (txMap.size() > 1) { //other ACID transactions are active
if (txMap.get(tx) <= ChronicleAcidIsolation.DIRTY_READ_OPTIMISTIC) {
b = this.compositeChronicleMap.get(cusip);
} else if (txMap.get(tx) >= ChronicleAcidIsolation.DIRTY_READ_INTOLERANT) {
System.out.println(
", @t="+System.currentTimeMillis()+
" Tx="+Thread.currentThread().toString().replaceAll(",",".")+
" inside chrAig.get() WAITING"+
" ,"
);
this.wait();
System.out.println(
" , @t="+System.currentTimeMillis()+
" Tx="+Thread.currentThread().toString().replaceAll(",",".")+
" inside chrAig.get() RESUMING"+
", "
);
b = this.compositeChronicleMap.get(cusip);
}
} else {
b = this.compositeChronicleMap.get(cusip);
}
System.out.println(
" , @t="+System.currentTimeMillis()+
" Tx="+Thread.currentThread().toString().replaceAll(",",".")+
" inside chrAig.get() DONE"+
" ,"
);
return b;
}
public ChronicleMap<String, Integer> getTransactionIsolationMap() {
return transactionIsolationMap;
}
public void setTransactionIsolationMap(ChronicleMap<String, Integer> txMap) {
this.transactionIsolationMap = txMap;
}
public ChronicleMap<String, BondVOInterface> getCompositeChronicleMap() {
return compositeChronicleMap;
}
public void setCompositeChronicleMap(ChronicleMap<String, BondVOInterface> ccm) {
this.compositeChronicleMap = ccm;
}
@Override
public synchronized void setTransactionIsolation(int level) throws SQLException {
this.getTransactionIsolationMap().put(
Thread.currentThread().toString(),
level
);
}
@Override
public synchronized int getTransactionIsolation() throws SQLException {
int iL = this.getTransactionIsolationMap().get(Thread.currentThread());
return(iL);
}
@Override
public synchronized void commit() throws SQLException {
System.out.println(
" ,@t="+System.currentTimeMillis()+
" Tx="+Thread.currentThread().toString().replaceAll(",",".")+
" chrAig.commit() BEGIN "+
", "
);
this.getTransactionIsolationMap().remove(Thread.currentThread().toString());
System.out.println(
" , @t="+System.currentTimeMillis()+
" Tx="+Thread.currentThread().toString().replaceAll(",",".")+
" chrAig.commit() END "+
", "
);
this.notifyAll();
System.out.println(
", @t="+System.currentTimeMillis()+
" Tx="+Thread.currentThread().toString().replaceAll(",",".")+
" chrAig.commit() complete notifyAll() to waiting Tx Threads "+
","
);
}
@Override
public synchronized void rollback() throws SQLException {
BondVOInterface priorBond = this.getCompositeChronicleMap().get("369604101");
if (priorBond != null) {
priorBond.setCoupon(3.50);
this.getCompositeChronicleMap().put("369604101", priorBond);
}
System.out.println(
" , @t="+System.currentTimeMillis()+
" Tx="+Thread.currentThread().toString().replaceAll(",",".")+
" chrAig.rollback() BEGIN "+
", "
);
this.getTransactionIsolationMap().remove(Thread.currentThread().toString());
System.out.println(
" ,@t="+System.currentTimeMillis()+
" Tx="+Thread.currentThread().toString().replaceAll(",",".")+
" chrAig.rollback() COMPLETE coupon=3.50"+
", "
);
this.notifyAll();
System.out.println(
", @t="+System.currentTimeMillis()+
" Tx="+Thread.currentThread().toString().replaceAll(",",".")+
" chrAig.rollback() completed notifyAll() to waiting Tx Threads"+
", "
);
}
// rest of these java.sql.Connection methods remain unimplemented ...
// pedantics are too
// intense. In the real world, even in Captital Markets 'dirty read' intolerance
// is likely the only isolation level Chronicle would accommodate.
@Override
public void setAutoCommit(boolean autoCommit) throws SQLException {
}
@Override
public boolean getAutoCommit() throws SQLException {
return false;
}
@Override
public void rollback(Savepoint savepoint) throws SQLException {
}
@Override
public void close() throws SQLException {
}
@Override
public Statement createStatement() throws SQLException {
return null;
}
@Override
public PreparedStatement prepareStatement(String sql) throws SQLException {
return null;
}
@Override
public CallableStatement prepareCall(String sql) throws SQLException {
return null;
}
@Override
public String nativeSQL(String sql) throws SQLException {
return null;
}
@Override
public boolean isClosed() throws SQLException {
return false;
}
@Override
public DatabaseMetaData getMetaData() throws SQLException {
return null;
}
@Override
public void setReadOnly(boolean readOnly) throws SQLException {
}
@Override
public boolean isReadOnly() throws SQLException {
return false;
}
@Override
public void setCatalog(String catalog) throws SQLException {
}
@Override
public String getCatalog() throws SQLException {
return null;
}
@Override
public SQLWarning getWarnings() throws SQLException {
return null;
}
@Override
public void clearWarnings() throws SQLException {
}
@Override
public Statement createStatement(int resultSetType, int resultSetConcurrency) throws SQLException {
return null;
}
@Override
public PreparedStatement prepareStatement(String sql, int resultSetType, int resultSetConcurrency) throws SQLException {
return null;
}
@Override
public CallableStatement prepareCall(String sql, int resultSetType, int resultSetConcurrency) throws SQLException {
return null;
}
@Override
public Map<String, Class<?>> getTypeMap() throws SQLException {
return null;
}
@Override
public void setTypeMap(Map<String, Class<?>> map) throws SQLException {
}
@Override
public void setHoldability(int holdability) throws SQLException {
}
@Override
public int getHoldability() throws SQLException {
return 0;
}
@Override
public Savepoint setSavepoint() throws SQLException {
return null;
}
@Override
public Savepoint setSavepoint(String name) throws SQLException {
return null;
}
@Override
public void releaseSavepoint(Savepoint savepoint) throws SQLException {
}
@Override
public Statement createStatement(int resultSetType, int resultSetConcurrency, int resultSetHoldability) throws SQLException {
return null;
}
@Override
public PreparedStatement prepareStatement(String sql, int resultSetType, int resultSetConcurrency, int resultSetHoldability) throws SQLException {
return null;
}
@Override
public CallableStatement prepareCall(String sql, int resultSetType, int resultSetConcurrency, int resultSetHoldability) throws SQLException {
return null;
}
@Override
public PreparedStatement prepareStatement(String sql, int autoGeneratedKeys) throws SQLException {
return null;
}
@Override
public PreparedStatement prepareStatement(String sql, int[] columnIndexes) throws SQLException {
return null;
}
@Override
public PreparedStatement prepareStatement(String sql, String[] columnNames) throws SQLException {
return null;
}
@Override
public Clob createClob() throws SQLException {
return null;
}
@Override
public Blob createBlob() throws SQLException {
return null;
}
@Override
public NClob createNClob() throws SQLException {
return null;
}
@Override
public SQLXML createSQLXML() throws SQLException {
return null;
}
@Override
public boolean isValid(int timeout) throws SQLException {
return false;
}
@Override
public void setClientInfo(String name, String value) throws SQLClientInfoException {
}
@Override
public void setClientInfo(Properties properties) throws SQLClientInfoException {
}
@Override
public String getClientInfo(String name) throws SQLException {
return null;
}
@Override
public Properties getClientInfo() throws SQLException {
return null;
}
@Override
public Array createArrayOf(String typeName, Object[] elements) throws SQLException {
return null;
}
@Override
public Struct createStruct(String typeName, Object[] attributes) throws SQLException {
return null;
}
@Override
public void setSchema(String schema) throws SQLException {
}
@Override
public String getSchema() throws SQLException {
return null;
}
@Override
public void abort(Executor executor) throws SQLException {
}
@Override
public void setNetworkTimeout(Executor executor, int milliseconds) throws SQLException {
}
@Override
public int getNetworkTimeout() throws SQLException {
return 0;
}
@Override
public <T> T unwrap(Class<T> iface) throws SQLException {
return null;
}
@Override
public boolean isWrapperFor(Class<?> iface) throws SQLException {
return false;
}
}
src/test/java/net/openhft/chronicle/map/fromdocs/acid/ChronicleStampedLock.pptx
BIN
View File
Binary file not shown.
src/test/java/net/openhft/chronicle/map/fromdocs/acid/ChronicleStampedLock=WANTED.png
BIN
View File
Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 142 KiB

src/test/java/net/openhft/chronicle/map/fromdocs/acid/ReadMe.adoc
+1
View File
@@ -0,0 +1 @@
image::ChronicleStampedLock=WANTED.png[]
src/test/java/net/openhft/chronicle/map/fromdocs/acid/exodus/DirtyReadOffender.java
+105
View File
@@ -0,0 +1,105 @@
package net.openhft.chronicle.map.fromdocs.acid.exodus;
import net.openhft.chronicle.map.ChronicleMap;
import net.openhft.chronicle.map.fromdocs.BondVOInterface;
import java.util.Scanner;
import java.util.concurrent.locks.StampedLock;
import static net.openhft.chronicle.values.Values.newNativeReference;
public class DirtyReadOffender implements Runnable {
private int isoLevel;
private ChronicleMap chm;
private StampedLock offHeapLock;
@Override
public void run() {
Scanner sc = new Scanner(System.in);
try {
BondVOInterface bond = newNativeReference(BondVOInterface.class);
this.chm.acquireUsing("369604101", bond);
System.out.println(
" @t=" + System.currentTimeMillis() +
" DirtyReadOffender sleeping 10 seconds "
);
Thread.sleep(10 * 1_000);
System.out.println(
" @t=" + System.currentTimeMillis() +
" DirtyReadOffender awakening "
);
/**
* ben.cotton@rutgers.edu ... anticipate Chronicle (www.OpenHFT.net)
* providing a j.u.c.l.StampedLock API for off-heap enthusiasts
*
* START
*
*/
long stamp = 0;
System.out.println(
" @t=" + System.currentTimeMillis() +
" DirtyReadOffender ACQUIRING offHeapLock.writeLock();"
);
while ((stamp = this.offHeapLock.writeLock()) == 0) {
;
}
System.out.println(
" @t=" + System.currentTimeMillis() +
" DirtyReadOffender ACQUIRED offHeapLock.writeLock();"
);
try {
double newCoupon = 3.5 + Math.random();
System.out.println(
" @t=" + System.currentTimeMillis() +
" DirtyReadOffender "+
" calling chm.put('369604101',"+newCoupon+") "
);
bond.setCoupon(newCoupon);
chm.put("369604101", bond);
System.out.println(
" @t=" + System.currentTimeMillis() +
" DirtyReadOffender coupon=["+
bond.getCoupon()+
"] written. "
);
} finally {
this.offHeapLock.unlockWrite(stamp);
System.out.println(
" @t=" + System.currentTimeMillis() +
" DirtyReadOffender called offHeapLock.unlockWrite(stamp);"
);
}
/**
* ben.cotton@rutgers.edu
*
* END
*
*/
} catch (Exception throwables) {
throwables.printStackTrace();
}
}
public ChronicleMap getCraig() {
return this.chm;
}
public void setCraig(ChronicleMap craig) {
this.chm = craig;
}
public int getIsoLevel() {
return isoLevel;
}
public void setIsoLevel(int isoLevel) {
this.isoLevel = isoLevel;
}
public void setStampedLock(StampedLock _sLock) {
this.offHeapLock = _sLock;
}
}
src/test/java/net/openhft/chronicle/map/fromdocs/acid/exodus/DirtyReadTolerance.java
+52
View File
@@ -0,0 +1,52 @@
package net.openhft.chronicle.map.fromdocs.acid.exodus;
import net.openhft.affinity.AffinitySupport;
import net.openhft.chronicle.map.ChronicleMap;
import net.openhft.chronicle.map.ChronicleMapBuilder;
import net.openhft.chronicle.map.fromdocs.BondVOInterface;
import java.io.File;
import java.io.IOException;
import java.util.concurrent.locks.StampedLock;
public class DirtyReadTolerance<K,V> {
public static void main(String args[]) throws Exception {
System.out.println("DirtyReadOffender,chrAig["+args[0]+"]coupon=3.50,DirtyReadVictim");
String isoLevel = args[0];
ChronicleMap<String, BondVOInterface> chm =
DirtyReadTolerance.offHeap(
"C:\\Users\\buddy\\dev\\shm\\OPERAND_CHRONICLE_MAP"
);
System.out.println(",Established chm,");
StampedLock sLock = new StampedLock();
DirtyReadOffender offender = new DirtyReadOffender();
offender.setCraig(chm);
offender.setStampedLock(sLock);
DirtyReadVictim victim = new DirtyReadVictim(isoLevel);
victim.setCraig(chm);
victim.setStampedLock(sLock);
(new Thread(offender)).start();
(new Thread(victim)).start();
}
static ChronicleMap<String, BondVOInterface> offHeap(String operand) throws IOException {
// ensure thread ids are globally unique.
AffinitySupport.setThreadId();
return ChronicleMapBuilder.of(String.class, BondVOInterface.class)
.entries(16)
.averageKeySize("123456789".length())
.createPersistedTo(
new File(
operand
// "C:\\Users\\buddy\\dev\\shm\\OPERAND_CHRONICLE_MAP"
)
);
//.create();
}
}
src/test/java/net/openhft/chronicle/map/fromdocs/acid/exodus/DirtyReadVictim.java
+115
View File
@@ -0,0 +1,115 @@
package net.openhft.chronicle.map.fromdocs.acid.exodus;
import net.openhft.chronicle.map.ChronicleMap;
import net.openhft.chronicle.map.fromdocs.BondVOInterface;
import net.openhft.chronicle.map.fromdocs.acid.ChronicleAcidIsolation;
import java.util.Scanner;
import java.util.concurrent.locks.StampedLock;
import static net.openhft.chronicle.values.Values.newNativeReference;
public class DirtyReadVictim implements Runnable {
private int isoLevel = ChronicleAcidIsolation.LOWEST_LATENCY;
private ChronicleMap chm;
private StampedLock offHeapLock;
DirtyReadVictim(String isoL) {
if (isoL.equals("DIRTY_READ_INTOLERANT"))
this.isoLevel = ChronicleAcidIsolation.DIRTY_READ_INTOLERANT;
else if (isoL.equals("DIRTY_READ_OPTIMISTIC"))
this.isoLevel = ChronicleAcidIsolation.DIRTY_READ_OPTIMISTIC;
}
@Override
public void run() {
Scanner sc = new Scanner(System.in);
try {
/**
* ben.cotton@rutgers.edu START
*/
Double coupon = 0.00;
BondVOInterface bond = newNativeReference(BondVOInterface.class);
long stamp = 0;
System.out.println(
" ,,@t=" + System.currentTimeMillis() +
" DirtyReadVictim CALLING offHeapLock.tryOptimisticRead()"
);
while ((stamp = this.offHeapLock.tryOptimisticRead()) == 0) {
;
}
System.out.println(
" ,,@t=" + System.currentTimeMillis() +
" DirtyReadVictim CALLED offHeapLock.tryOptimisticRead()"
);
try {
chm.acquireUsing("369604101", bond);
System.out.println(
" ,,@t=" + System.currentTimeMillis() +
" DirtyReadVictim calling chm.get('369604101').getCoupon()"
);
bond = (BondVOInterface) chm.get("369604101");
coupon = bond.getCoupon();
System.out.println(
" ,,@t=" + System.currentTimeMillis() +
" DirtyReadVictim coupon=[" +coupon+"] read."
);
System.out.println(
" ,,@t=" + System.currentTimeMillis() +
" DirtyReadVictim sleeping 60 seconds"
);
Thread.sleep( 60_000);
} finally {
if (this.offHeapLock.validate(stamp)) {
System.out.println(
" ,,@t=" + System.currentTimeMillis() +
" DirtyReadVictim OPTIMISTICALLY_READ coupon=" +
coupon + " "
);
} else {
System.out.println(
" ,,@t=" + System.currentTimeMillis() +
" DirtyReadVictim FAILED offHeapLock.validate(stamp) " +
" must apply PESSIMISTIC_POLICY (dirty read endured)" +
" coupon=[" + coupon + "] is *DIRTY*. "
);
}
}
/**
* ben.cotton@rutgers.edu END
*/
System.out.println(
" ,,@t=" + System.currentTimeMillis() +
" DirtyReadVictim got() coupon=" +
coupon + " "
);
System.out.println(
" ,,@t=" + System.currentTimeMillis() +
" DirtyReadVictim COMMITTED"
);
} catch (Exception throwables) {
throwables.printStackTrace();
}
}
public ChronicleMap getCraig() {
return this.chm;
}
public void setCraig(ChronicleMap craig) {
this.chm = craig;
}
public void setStampedLock(StampedLock _sLock) {
this.offHeapLock = _sLock;
}
}
src/test/java/net/openhft/chronicle/map/fromdocs/acid/genesis/DirtyReadOffender.java
+105
View File
@@ -0,0 +1,105 @@
package net.openhft.chronicle.map.fromdocs.acid.genesis;
import net.openhft.chronicle.map.ChronicleMap;
import net.openhft.chronicle.map.fromdocs.BondVOInterface;
import java.util.Scanner;
import java.util.concurrent.locks.StampedLock;
import static net.openhft.chronicle.values.Values.newNativeReference;
public class DirtyReadOffender implements Runnable {
private int isoLevel;
private ChronicleMap chm;
private StampedLock offHeapLock;
@Override
public void run() {
Scanner sc = new Scanner(System.in);
try {
BondVOInterface bond = newNativeReference(BondVOInterface.class);
this.chm.acquireUsing("369604101", bond);
System.out.println(
" @t=" + System.currentTimeMillis() +
" DirtyReadOffender sleeping 60 seconds "
);
Thread.sleep(60 * 1_000);
System.out.println(
" @t=" + System.currentTimeMillis() +
" DirtyReadOffender awakening "
);
/**
* ben.cotton@rutgers.edu ... anticipate Chronicle (www.OpenHFT.net)
* providing a j.u.c.l.StampedLock API for off-heap enthusiasts
*
* START
*
*/
long stamp = 0;
System.out.println(
" @t=" + System.currentTimeMillis() +
" DirtyReadOffender ACQUIRING offHeapLock.writeLock();"
);
while ((stamp = this.offHeapLock.writeLock()) == 0) {
;
}
System.out.println(
" @t=" + System.currentTimeMillis() +
" DirtyReadOffender ACQUIRED offHeapLock.writeLock();"
);
try {
double newCoupon = 3.5 + Math.random();
System.out.println(
" @t=" + System.currentTimeMillis() +
" DirtyReadOffender "+
" calling chm.put('369604101',"+newCoupon+") "
);
bond.setCoupon(newCoupon);
chm.put("369604101", bond);
System.out.println(
" @t=" + System.currentTimeMillis() +
" DirtyReadOffender coupon=["+
bond.getCoupon()+
"] written. "
);
} finally {
this.offHeapLock.unlockWrite(stamp);
System.out.println(
" @t=" + System.currentTimeMillis() +
" DirtyReadOffender called offHeapLock.unlockWrite(stamp);"
);
}
/**
* ben.cotton@rutgers.edu
*
* END
*
*/
} catch (Exception throwables) {
throwables.printStackTrace();
}
}
public ChronicleMap getCraig() {
return this.chm;
}
public void setCraig(ChronicleMap craig) {
this.chm = craig;
}
public int getIsoLevel() {
return isoLevel;
}
public void setIsoLevel(int isoLevel) {
this.isoLevel = isoLevel;
}
public void setStampedLock(StampedLock _sLock) {
this.offHeapLock = _sLock;
}
}
src/test/java/net/openhft/chronicle/map/fromdocs/acid/genesis/DirtyReadTolerance.java
+52
View File
@@ -0,0 +1,52 @@
package net.openhft.chronicle.map.fromdocs.acid.genesis;
import net.openhft.affinity.AffinitySupport;
import net.openhft.chronicle.map.ChronicleMap;
import net.openhft.chronicle.map.ChronicleMapBuilder;
import net.openhft.chronicle.map.fromdocs.BondVOInterface;
import java.io.File;
import java.io.IOException;
import java.util.concurrent.locks.StampedLock;
public class DirtyReadTolerance<K,V> {
public static void main(String args[]) throws Exception {
System.out.println("DirtyReadOffender,chrAig["+args[0]+"]coupon=3.50,DirtyReadVictim");
String isoLevel = args[0];
ChronicleMap<String, BondVOInterface> chm =
DirtyReadTolerance.offHeap(
"C:\\Users\\buddy\\dev\\shm\\OPERAND_CHRONICLE_MAP"
);
System.out.println(",Established chm,");
StampedLock sLock = new StampedLock();
DirtyReadOffender offender = new DirtyReadOffender();
offender.setCraig(chm);
offender.setStampedLock(sLock);
DirtyReadVictim victim = new DirtyReadVictim(isoLevel);
victim.setCraig(chm);
victim.setStampedLock(sLock);
(new Thread(offender)).start();
(new Thread(victim)).start();
}
static ChronicleMap<String, BondVOInterface> offHeap(String operand) throws IOException {
// ensure thread ids are globally unique.
AffinitySupport.setThreadId();
return ChronicleMapBuilder.of(String.class, BondVOInterface.class)
.entries(16)
.averageKeySize("123456789".length())
.createPersistedTo(
new File(
operand
// "C:\\Users\\buddy\\dev\\shm\\OPERAND_CHRONICLE_MAP"
)
);
//.create();
}
}
src/test/java/net/openhft/chronicle/map/fromdocs/acid/genesis/DirtyReadVictim.java
+115
View File
@@ -0,0 +1,115 @@
package net.openhft.chronicle.map.fromdocs.acid.genesis;
import net.openhft.chronicle.map.ChronicleMap;
import net.openhft.chronicle.map.fromdocs.BondVOInterface;
import net.openhft.chronicle.map.fromdocs.acid.ChronicleAcidIsolation;
import java.util.Scanner;
import java.util.concurrent.locks.StampedLock;
import static net.openhft.chronicle.values.Values.newNativeReference;
public class DirtyReadVictim implements Runnable {
private int isoLevel = ChronicleAcidIsolation.LOWEST_LATENCY;
private ChronicleMap chm;
private StampedLock offHeapLock;
DirtyReadVictim(String isoL) {
if (isoL.equals("DIRTY_READ_INTOLERANT"))
this.isoLevel = ChronicleAcidIsolation.DIRTY_READ_INTOLERANT;
else if (isoL.equals("DIRTY_READ_OPTIMISTIC"))
this.isoLevel = ChronicleAcidIsolation.DIRTY_READ_OPTIMISTIC;
}
@Override
public void run() {
Scanner sc = new Scanner(System.in);
try {
/**
* ben.cotton@rutgers.edu START
*/
Double coupon = 0.00;
BondVOInterface bond = newNativeReference(BondVOInterface.class);
long stamp = 0;
System.out.println(
" ,,@t=" + System.currentTimeMillis() +
" DirtyReadVictim CALLING offHeapLock.tryOptimisticRead()"
);
while ((stamp = this.offHeapLock.tryOptimisticRead()) == 0) {
;
}
System.out.println(
" ,,@t=" + System.currentTimeMillis() +
" DirtyReadVictim CALLED offHeapLock.tryOptimisticRead()"
);
try {
chm.acquireUsing("369604101", bond);
System.out.println(
" ,,@t=" + System.currentTimeMillis() +
" DirtyReadVictim calling chm.get('369604101').getCoupon()"
);
bond = (BondVOInterface) chm.get("369604101");
coupon = bond.getCoupon();
System.out.println(
" ,,@t=" + System.currentTimeMillis() +
" DirtyReadVictim coupon=[" +coupon+"] read."
);
System.out.println(
" ,,@t=" + System.currentTimeMillis() +
" DirtyReadVictim sleeping 10 seconds"
);
Thread.sleep( 10_000);
} finally {
if (this.offHeapLock.validate(stamp)) {
System.out.println(
" ,,@t=" + System.currentTimeMillis() +
" DirtyReadVictim OPTIMISTICALLY_READ coupon=" +
coupon + " "
);
} else {
System.out.println(
" ,,@t=" + System.currentTimeMillis() +
" DirtyReadVictim FAILED offHeapLock.validate(stamp) " +
" must apply PESSIMISTIC_POLICY (dirty read endured)" +
" coupon=[" + coupon + "] is *DIRTY*. "
);
}
}
/**
* ben.cotton@rutgers.edu END
*/
System.out.println(
" ,,@t=" + System.currentTimeMillis() +
" DirtyReadVictim got() coupon=" +
coupon + " "
);
System.out.println(
" ,,@t=" + System.currentTimeMillis() +
" DirtyReadVictim COMMITTED"
);
} catch (Exception throwables) {
throwables.printStackTrace();
}
}
public ChronicleMap getCraig() {
return this.chm;
}
public void setCraig(ChronicleMap craig) {
this.chm = craig;
}
public void setStampedLock(StampedLock _sLock) {
this.offHeapLock = _sLock;
}
}
src/test/java/net/openhft/chronicle/map/fromdocs/acid/revelations/ChronicleStampedLock.java
+296
View File
@@ -0,0 +1,296 @@
package net.openhft.chronicle.map.fromdocs.acid.revelations;
import net.openhft.affinity.AffinitySupport;
import net.openhft.chronicle.core.values.LongValue;
import net.openhft.chronicle.map.ChronicleMap;
import net.openhft.chronicle.map.ChronicleMapBuilder;
import net.openhft.chronicle.values.Values;
import java.io.File;
import java.io.IOException;
import java.util.concurrent.locks.StampedLock;
import static net.openhft.chronicle.values.Values.newNativeReference;
/**
* ben.cotton@rutgers.edu
* <p>
* A totally hacked impl, awaiting OpenHFT's official ChronicleStampedLock API
* <p>
* A usable impl will more properly belong in the Chronicle-Algorithms repo
*/
public class ChronicleStampedLock extends StampedLock {
ChronicleMap<String, ChronicleStampedLockVOInterface> chm;
ChronicleMap<String, LongValue> chmR;
ChronicleStampedLockVOInterface offHeapLock =
newNativeReference(ChronicleStampedLockVOInterface.class);
ChronicleStampedLockVOInterface lastWriterT =
newNativeReference(ChronicleStampedLockVOInterface.class);
LongValue readLockHolderCount = Values.newNativeReference(LongValue.class);
ChronicleStampedLock(String chronicelStampedLockLocality) {
try {
chm = offHeapLock(chronicelStampedLockLocality);
chmR = offHeapLockReaderCount(chronicelStampedLockLocality+"=ReaderCount");
chm.acquireUsing("Stamp ", offHeapLock);
chm.acquireUsing("LastWriterTime ", lastWriterT);
chmR.acquireUsing("ReaderCount ", readLockHolderCount);
//readLockHolderCount.setValue(0L);
System.out.println(
" ,@t=" + System.currentTimeMillis() +
" ChronicleStampedLock constructed" +
","
);
} catch (IOException e) {
e.printStackTrace();
}
}
@Override
public long tryOptimisticRead() {
long l = 0;
offHeapLock = chm.get("Stamp ");
//do {
l = (offHeapLock = chm.get("Stamp ")).getEntryLockState();
// System.out.println(
// " ,,@t=" + System.currentTimeMillis() +
// " ChronicleStampedLock waiting for unlockWrite()... " +
// ""
// );
// try {
// Thread.sleep(1000);
// } catch (InterruptedException e) {
// e.printStackTrace();
// }
// } while ( /*l < 0L*/); // do not loop! @vgrazi confirms that in JCA rebooted app.
offHeapLock.setEntryLockState(0L);
long t = System.currentTimeMillis();
chm.put("Stamp ", offHeapLock);
System.out.println(
" ,@t=" + System.currentTimeMillis() +
" ChronicleStampedLock saw stamp=["+l+"]"+
" tryOptmisticRead() returning stamp=" +
t +
","
);
return (t);
}
@Override
public boolean validate(long stamp) {
offHeapLock = chm.get("Stamp ");
lastWriterT = chm.get("LastWriterTime ");
boolean ret = false;
if (lastWriterT.getEntryLockState() > stamp || offHeapLock.getEntryLockState() <0L) {
ret = Boolean.FALSE;
} else {
ret = Boolean.TRUE;
}
System.out.println(
" ,@t=" + System.currentTimeMillis() +
" ChronicleStampedLock validate(" +
stamp + ") returned =[" +
ret + "] " +
","
);
System.out.println(
" ,@t=" + System.currentTimeMillis() +
" ChronicleStampedLock LastWriterT=[" +
lastWriterT.getEntryLockState() + "] " +
","
);
return ret;
}
@Override
public long tryWriteLock() {
long l = 0;
offHeapLock = chm.get("Stamp ");
lastWriterT = chm.get("LastWriterTime ");
do {
System.out.println(
" ,@t=" + System.currentTimeMillis() +
" ChronicleStampedLock tryWriterLock() WAITING " +
" on offHeapLock.unlock(" +
offHeapLock.getEntryLockState() +
") ,"
);
offHeapLock = chm.get("Stamp ");
l = offHeapLock.getEntryLockState();
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
} while (l != 0L || readLockHolderCount.getValue() > 0);
System.out.println(
" ,@t=" + System.currentTimeMillis() +
" ChronicleStampedLock tryWriterLock() RESUMING " +
" ,"
);
long t = System.currentTimeMillis();
try {
Thread.sleep((long) (Math.random() * 1000)); //sleep up to 1 second
} catch (InterruptedException e) {
e.printStackTrace();
}
offHeapLock.setEntryLockState(-t); //negative ==> Writer holds StampedLock
lastWriterT.setEntryLockState(t);
chm.put("Stamp ", offHeapLock);
chm.put("LastWriterTime ", lastWriterT);
System.out.println(
" ,@t=" + t +
" ChronicleStampedLock tryWriteLock() returned stamp=" +
offHeapLock.getEntryLockState() +
","
);
return (offHeapLock.getEntryLockState());
}
@Override
public long tryReadLock() {
long l = 0;
offHeapLock = chm.get("Stamp ");
readLockHolderCount = chmR.get("ReaderCount ");
do {
System.out.println(
" ,@t=" + System.currentTimeMillis() +
" ChronicleStampedLock tryReadLock() WAITING on Writer.unlock(" +
offHeapLock.getEntryLockState() +
") ,"
);
l = (offHeapLock = chm.get("Stamp ")).getEntryLockState();
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
} while (l < 0L);
try {
Thread.sleep((long) (Math.random() * 2000)); //sleep up to 1 second
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(
" ,@t=" + System.currentTimeMillis() +
" ChronicleStampedLock tryReadLock() RESUMING " +
//" readerCount=[" + readerCount.getReaderCount() + "]" +
" readerCount=[" + readLockHolderCount.getVolatileValue() + "]" +
" BEFORE addAtomic(1) "+
","
);
readLockHolderCount.addAtomicValue(1);
try {
Thread.sleep((long) (Math.random() * 1000)); //sleep up to 1 second
} catch (InterruptedException e) {
e.printStackTrace();
}
offHeapLock.setEntryLockState(
readLockHolderCount.getVolatileValue()
); //negative ==> Writer holds StampedLock
chm.put("Stamp ", offHeapLock);
System.out.println(
" ,@t=" + System.currentTimeMillis() +
" ChronicleStampedLock tryReadLock() returned stamp=" +
offHeapLock.getEntryLockState() +
" readerCount=[" + readLockHolderCount.getVolatileValue() + "]" +
" AFTER addAtomic(1) "+
","
);
chmR.put("ReaderCount ", readLockHolderCount);
return (offHeapLock.getEntryLockState());
}
@Override
public void unlockRead(long stamp) {
offHeapLock = chm.get("Stamp ");
readLockHolderCount = chmR.get("ReaderCount ");
System.out.println(
" ,@t=" + System.currentTimeMillis() +
" ChronicleStampedLock unlockRead(" + stamp + ") unlocking.." +
"ReaderCount=[" + readLockHolderCount.getVolatileValue() + "]" +
" BEFORE addAtomic(-1) "+
","
);
readLockHolderCount.addAtomicValue(-1);
System.out.println(
" ,@t=" + System.currentTimeMillis() +
" ChronicleStampedLock unlockRead(" + stamp + ") unlocking.." +
"ReaderCount=[" + readLockHolderCount.getVolatileValue() + "]" +
" AFTER addAtomic(-1) "+
","
);
offHeapLock.setEntryLockState(readLockHolderCount.getVolatileValue());
System.out.println(
" ,@t=" + System.currentTimeMillis() +
"offHeapLock=[" + offHeapLock.getEntryLockState() + "]" +
","
);
chm.put("Stamp ", offHeapLock);
chmR.put("ReaderCount ", readLockHolderCount);
}
@Override
public void unlockWrite(long stamp) {
System.out.println(
" ,@t=" + System.currentTimeMillis() +
" ChronicleStampedLock unlockWrite(" + stamp + ") unlocking..,"
);
offHeapLock = chm.get("Stamp ");
offHeapLock.setEntryLockState(0L);
chm.put("Stamp ", offHeapLock);
System.out.println(
" ,@t=" + System.currentTimeMillis() +
" ChronicleStampedLock unlockWrite(" +
chm.get("Stamp ").getEntryLockState() +
") unlocked. set to Zero" +
","
);
}
static ChronicleMap<String, ChronicleStampedLockVOInterface> offHeapLock(String operand)
throws IOException {
AffinitySupport.setThreadId();
return ChronicleMapBuilder.of(String.class, ChronicleStampedLockVOInterface.class)
.entries(16)
.averageKeySize("123456789".length())
.createPersistedTo(
new File(
operand
)
);
}
static ChronicleMap<String, LongValue> offHeapLockReaderCount(String operand)
throws IOException {
AffinitySupport.setThreadId();
return ChronicleMapBuilder.of(String.class, LongValue.class)
.entries(16)
.averageKeySize("123456789".length())
.createPersistedTo(
new File(
operand
)
);
}
}
src/test/java/net/openhft/chronicle/map/fromdocs/acid/revelations/ChronicleStampedLockVOInterface.java
+38
View File
@@ -0,0 +1,38 @@
/*
* Copyright 2012-2018 Chronicle Map Contributors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package net.openhft.chronicle.map.fromdocs.acid.revelations;
import net.openhft.chronicle.values.Array;
import net.openhft.chronicle.values.Group;
import net.openhft.chronicle.values.MaxUtf8Length;
interface ChronicleStampedLockVOInterface {
@Group(0)
long getEntryLockState();
void setEntryLockState(long entryLockState);
//
// @Group(1)
// long getReaderCount();
//
// void setReaderCount(long rc); /* time in millis */
//
// long addAtomicReaderCount(long toAdd);
//
}
src/test/java/net/openhft/chronicle/map/fromdocs/acid/revelations/DirtyReadIntolerant.java
+101
View File
@@ -0,0 +1,101 @@
package net.openhft.chronicle.map.fromdocs.acid.revelations;
import net.openhft.chronicle.map.ChronicleMap;
import net.openhft.chronicle.map.fromdocs.BondVOInterface;
import java.util.concurrent.locks.StampedLock;
import static net.openhft.chronicle.values.Values.newNativeReference;
public class DirtyReadIntolerant {
public static void main(String args[]) {
try {
String isoLevel = args[0];
long sleepMock = Long.parseLong(args[1]);
long holdTime = Long.parseLong(args[2]);
/**
* ben.cotton@rutgers.edu START
*/
ChronicleMap<String, BondVOInterface> chm =
DirtyReadTolerance.offHeap(
args[3]
+ "OPERAND_CHRONICLE_MAP"
);
Double coupon = 0.00;
BondVOInterface bond = newNativeReference(BondVOInterface.class);
//BondVOInterface cslMock = newNativeReference(BondVOInterface.class); //mock'd
long stamp = 0;
System.out.println(
" ,,@t=" + System.currentTimeMillis() +
" DirtyReadIntolerant ENTERING offHeapLock.tryReadLock()"
);
StampedLock offHeapLock =
new ChronicleStampedLock(
args[3]
+ "OPERAND_ChronicleStampedLock"
);
System.out.println(
" ,,@t=" + System.currentTimeMillis() +
" DirtyReadIntolerant sleeping " + sleepMock + " seconds"
);
Thread.sleep(sleepMock * 1_000);
while ((stamp = offHeapLock.tryReadLock()) < 0) {
;
}
System.out.println(
" ,,@t=" + System.currentTimeMillis() +
" DirtyReadIntolerant ENTERED offHeapLock.tryReadLock() " +
" stamp=[" +
stamp +
"]"
);
try {
chm.acquireUsing("369604101", bond);
//chm.acquireUsing("Offender ", cslMock); //mock'd
System.out.println(
" ,,@t=" + System.currentTimeMillis() +
" DirtyReadIntolerant calling chm.get('369604101').getCoupon()"
);
bond = (BondVOInterface) chm.get("369604101");
coupon = bond.getCoupon();
System.out.println(
" ,,@t=" + System.currentTimeMillis() +
" DirtyReadIntolerant coupon=[" + coupon + "] read."
);
System.out.println(
" ,,@t=" + System.currentTimeMillis() +
" DirtyReadIntolerant sleeping " + holdTime + " seconds"
);
Thread.sleep(holdTime * 1_000);
System.out.println(
" ,,@t=" + System.currentTimeMillis() +
" DirtyReadIntolerant awakening"
);
} finally {
offHeapLock.unlockRead(stamp);
System.out.println(
" ,,@t=" + System.currentTimeMillis() +
" DirtyReadIntolerant offHeapLock.unlockRead(" +
stamp + ") completed."
);
}
/**
* ben.cotton@rutgers.edu END
*/
System.out.println(
" ,,@t=" + System.currentTimeMillis() +
" DirtyReadIntolerant got() coupon=" +
coupon + " "
);
System.out.println(
" ,,@t=" + System.currentTimeMillis() +
" DirtyReadIntolerant COMMITTED"
);
} catch (Exception throwables) {
throwables.printStackTrace();
}
}
}
src/test/java/net/openhft/chronicle/map/fromdocs/acid/revelations/DirtyReadOffender.java
+111
View File
@@ -0,0 +1,111 @@
package net.openhft.chronicle.map.fromdocs.acid.revelations;
import net.openhft.chronicle.map.ChronicleMap;
import net.openhft.chronicle.map.fromdocs.BondVOInterface;;
import java.util.Scanner;
import java.util.concurrent.locks.StampedLock;
import static net.openhft.chronicle.values.Values.newNativeReference;
public class DirtyReadOffender {
public static void main(String args[]) {
Scanner sc = new Scanner(System.in);
try {
String isoLevel = args[0];
long sleepT = Long.parseLong(args[1]);
long holdTime = Long.parseLong(args[2]);
ChronicleMap<String, BondVOInterface> chm =
DirtyReadTolerance.offHeap(
args[3]
+ "OPERAND_CHRONICLE_MAP"
);
System.out.println(
" @t=" + System.currentTimeMillis() +
" DirtyReadOffender established chm "
);
StampedLock offHeapLock = new ChronicleStampedLock(
args[3]
+ "OPERAND_ChronicleStampedLock"
);
BondVOInterface bond = newNativeReference(BondVOInterface.class);
//BondVOInterface cslMock = newNativeReference(BondVOInterface.class);
chm.acquireUsing("369604101", bond);
//chm.acquireUsing("Offender ", cslMock); // mock ChronicleStampLock
System.out.println(
" @t=" + System.currentTimeMillis() +
" DirtyReadOffender sleeping " + sleepT + " seconds "
);
Thread.sleep(sleepT * 1_000);
System.out.println(
" @t=" + System.currentTimeMillis() +
" DirtyReadOffender awakening "
);
/**
* ben.cotton@rutgers.edu ... anticipate Chronicle (www.OpenHFT.net)
* providing a j.u.c.l.StampedLock API for off-heap enthusiasts
*
* START
*
*/
long stamp = 0;
System.out.println(
" @t=" + System.currentTimeMillis() +
" DirtyReadOffender ACQUIRING offHeapLock.writeLock();"
);
while ((stamp = offHeapLock.tryWriteLock()) == 0) {
;
}
System.out.println(
" @t=" + System.currentTimeMillis() +
" DirtyReadOffender ACQUIRED offHeapLock.writeLock();"
);
try {
double newCoupon = 3.5 + Math.random();
System.out.println(
" @t=" + System.currentTimeMillis() +
" DirtyReadOffender " +
" calling chm.put('369604101'," + newCoupon + ") "
);
bond.setCoupon(newCoupon);
chm.put("369604101", bond);
//cslMock.setEntryLockState(System.currentTimeMillis()); //mock'd
// chm.put("Offender ",cslMock); //mock'd
System.out.println(
" @t=" + System.currentTimeMillis() +
" DirtyReadOffender coupon=[" +
bond.getCoupon() +
"] written. "
);
} finally {
System.out.println(
" @t=" + System.currentTimeMillis() +
" DirtyReadOffender sleeping " + holdTime + " seconds "
);
Thread.sleep(holdTime * 1_000);
offHeapLock.unlockWrite(stamp);
System.out.println(
" @t=" + System.currentTimeMillis() +
" DirtyReadOffender called " +
"offHeapLock.unlockWrite(" + stamp + ");"
);
}
/**
* ben.cotton@rutgers.edu
*
* END
*
*/
} catch (Exception throwables) {
throwables.printStackTrace();
} finally {
System.out.println(
" ,,@t=" + System.currentTimeMillis() +
" DirtyReadOffender COMMITTED"
);
}
}
}
src/test/java/net/openhft/chronicle/map/fromdocs/acid/revelations/DirtyReadTolerance.java
+45
View File
@@ -0,0 +1,45 @@
package net.openhft.chronicle.map.fromdocs.acid.revelations;
import net.openhft.affinity.AffinitySupport;
import net.openhft.chronicle.map.ChronicleMap;
import net.openhft.chronicle.map.ChronicleMapBuilder;
import net.openhft.chronicle.map.fromdocs.BondVOInterface;
import net.openhft.chronicle.map.fromdocs.acid.exodus.DirtyReadOffender;
import net.openhft.chronicle.map.fromdocs.acid.exodus.DirtyReadVictim;
import java.io.File;
import java.io.IOException;
import java.util.concurrent.locks.StampedLock;
public class DirtyReadTolerance<K,V> {
public static void main(String args[]) throws Exception {
System.out.println("DirtyReadOffender,chrAig["+args[0]+"]coupon=3.50,DirtyReadVictim");
String isoLevel = args[0];
ChronicleMap<String, BondVOInterface> chm =
DirtyReadTolerance.offHeap(
args[2]
+ "OPERAND_CHRONICLE_MAP"
);
System.out.println(",Established chm,");
}
static ChronicleMap<String, BondVOInterface> offHeap(String operand) throws IOException {
// ensure thread ids are globally unique.
AffinitySupport.setThreadId();
return ChronicleMapBuilder.of(String.class, BondVOInterface.class)
.entries(16)
.averageKeySize("123456789".length())
.createPersistedTo(
new File(
operand
// "C:\\Users\\buddy\\dev\\shm\\OPERAND_CHRONICLE_MAP"
)
);
//.create();
}
}
src/test/java/net/openhft/chronicle/map/fromdocs/acid/revelations/DirtyReadVictim.java
+125
View File
@@ -0,0 +1,125 @@
package net.openhft.chronicle.map.fromdocs.acid.revelations;
import net.openhft.chronicle.map.ChronicleMap;
import net.openhft.chronicle.map.fromdocs.BondVOInterface;
import net.openhft.chronicle.map.fromdocs.acid.ChronicleAcidIsolation;
import java.util.concurrent.locks.StampedLock;
import static net.openhft.chronicle.values.Values.newNativeReference;
public class DirtyReadVictim {
public static void main(String args[]) {
try {
String isoLevel = args[0];
long sleepMock = Long.parseLong(args[1]);
long holdTime = Long.parseLong(args[2]);
/**
* ben.cotton@rutgers.edu START
*/
ChronicleMap<String, BondVOInterface> chm =
DirtyReadTolerance.offHeap(
args[3]
+ "OPERAND_CHRONICLE_MAP"
);
Double coupon = 0.00;
BondVOInterface bond = newNativeReference(BondVOInterface.class);
//BondVOInterface cslMock = newNativeReference(BondVOInterface.class); //mock'd
long stamp = 0;
System.out.println(
" ,,@t=" + System.currentTimeMillis() +
" DirtyReadVictim ENTERING offHeapLock.tryOptimisticRead()"
);
StampedLock offHeapLock =
new ChronicleStampedLock(
args[3]
+ "OPERAND_ChronicleStampedLock"
);
System.out.println(
" ,,@t=" + System.currentTimeMillis() +
" DirtyReadVictim sleeping " + sleepMock + " seconds"
);
Thread.sleep(sleepMock * 1_000);
while ((stamp = offHeapLock.tryOptimisticRead()) < 0) {
System.out.println(
" ,,@t=" + System.currentTimeMillis() +
" DirtyReadVictim waiting for unlockWrite()... " +
""
);
Thread.sleep(1000);
}
System.out.println(
" ,,@t=" + System.currentTimeMillis() +
" DirtyReadVictim ENTERED offHeapLock.tryOptimisticRead() " +
" stamp=[" +
stamp +
"]"
);
try {
chm.acquireUsing("369604101", bond);
//chm.acquireUsing("Offender ", cslMock); //mock'd
System.out.println(
" ,,@t=" + System.currentTimeMillis() +
" DirtyReadVictim calling chm.get('369604101').getCoupon()"
);
bond = (BondVOInterface) chm.get("369604101");
coupon = bond.getCoupon();
System.out.println(
" ,,@t=" + System.currentTimeMillis() +
" DirtyReadVictim coupon=[" + coupon + "] read."
);
System.out.println(
" ,,@t=" + System.currentTimeMillis() +
" DirtyReadVictim sleeping " + holdTime + " seconds"
);
Thread.sleep(holdTime * 1_000);
System.out.println(
" ,,@t=" + System.currentTimeMillis() +
" DirtyReadVictim awakening"
);
} finally {
if (offHeapLock.validate(stamp)) {
System.out.println(
" ,,@t=" + System.currentTimeMillis() +
" DirtyReadVictim OPTIMISTICALLY_READ coupon=" +
coupon + " " +
"stamp = [" + stamp + "]"
);
} else {
System.out.println(
" ,,@t=" + System.currentTimeMillis() +
" DirtyReadVictim FAILED offHeapLock.validate(stamp) " +
" must apply PESSIMISTIC_POLICY (dirty read endured)" +
" coupon=[" + coupon + "] is *DIRTY*. " +
" stamp = [" + stamp + "]"
);
}
System.out.println(
" ,,@t=" + System.currentTimeMillis() +
" DirtyReadVictim EXITED offHeapLock.tryOptimisticRead()"
);
}
/**
* ben.cotton@rutgers.edu END
*/
System.out.println(
" ,,@t=" + System.currentTimeMillis() +
" DirtyReadVictim got() coupon=" +
coupon + " "
);
System.out.println(
" ,,@t=" + System.currentTimeMillis() +
" DirtyReadVictim COMMITTED"
);
} catch (Exception throwables) {
throwables.printStackTrace();
}
}
}
src/test/java/net/openhft/chronicle/map/fromdocs/acid/revelations/DirtyReadVictimTest.java
+103
View File
@@ -0,0 +1,103 @@
package net.openhft.chronicle.map.fromdocs.acid.revelations;
import net.openhft.chronicle.map.ChronicleMap;
import net.openhft.chronicle.map.fromdocs.BondVOInterface;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import static net.openhft.chronicle.values.Values.newNativeReference;
import static org.junit.Assert.*;
public class DirtyReadVictimTest {
@Before
public void setUp() throws Exception {
}
@After
public void tearDown() throws Exception {
}
@Test
public void main() {
try {
/**
* ben.cotton@rutgers.edu START
*/
ChronicleMap<String, BondVOInterface> chm =
DirtyReadTolerance.offHeap(
"C:\\Users\\buddy\\dev\\shm\\OPERAND_CHRONICLE_MAP"
);
Double coupon = 0.00;
BondVOInterface bond = newNativeReference(BondVOInterface.class);
long stamp = 0;
System.out.println(
" ,,@t=" + System.currentTimeMillis() +
" DirtyReadVictim CALLING offHeapLock.tryOptimisticRead()"
);
ChronicleStampedLock offHeapLock = new ChronicleStampedLock(
"C:\\Users\\buddy\\dev\\shm\\OPERAND_ChronicleStampedLock"
);
while ((stamp = offHeapLock.tryOptimisticRead()) == 0) {
;
}
System.out.println(
" ,,@t=" + System.currentTimeMillis() +
" DirtyReadVictim CALLED offHeapLock.tryOptimisticRead()"
);
try {
chm.acquireUsing("369604101", bond);
System.out.println(
" ,,@t=" + System.currentTimeMillis() +
" DirtyReadVictim calling chm.get('369604101').getCoupon()"
);
bond = (BondVOInterface) chm.get("369604101");
coupon = bond.getCoupon();
System.out.println(
" ,,@t=" + System.currentTimeMillis() +
" DirtyReadVictim coupon=[" + coupon + "] read."
);
System.out.println(
" ,,@t=" + System.currentTimeMillis() +
" DirtyReadVictim sleeping 10 seconds"
);
Thread.sleep(10_000);
} finally {
if (offHeapLock.validate(stamp)) {
System.out.println(
" ,,@t=" + System.currentTimeMillis() +
" DirtyReadVictim OPTIMISTICALLY_READ coupon=" +
coupon + " "
);
// THIS Test will pass when ChronicleStampedLock is GA
Assert.assertFalse(Boolean.TRUE);
} else {
System.out.println(
" ,,@t=" + System.currentTimeMillis() +
" DirtyReadVictim FAILED offHeapLock.validate(stamp) " +
" must apply PESSIMISTIC_POLICY (dirty read endured)" +
" coupon=[" + coupon + "] is *DIRTY*. "
);
// THIS Test will execute pass when ChronicleStampedLock is GA
Assert.assertFalse(Boolean.FALSE);
}
}
/**
* ben.cotton@rutgers.edu END
*/
System.out.println(
" ,,@t=" + System.currentTimeMillis() +
" DirtyReadVictim got() coupon=" +
coupon + " "
);
System.out.println(
" ,,@t=" + System.currentTimeMillis() +
" DirtyReadVictim COMMITTED"
);
} catch (Exception throwables) {
throwables.printStackTrace();
}
}
}
src/test/java/net/openhft/chronicle/map/fromdocs/acid/revelations/Weyuker,E.pdf
BIN
View File
Binary file not shown.
src/test/java/net/openhft/chronicle/map/fromdocs/acid/revelations/clrs=Algorithms.pdf
BIN
View File
Binary file not shown.
src/test/java/net/openhft/chronicle/map/fromdocs/pingpong_latency/PingPongLockLeft.java
+64 -64
View File
@@ -32,68 +32,68 @@ import static net.openhft.chronicle.algo.bytes.Access.checkedBytesStoreAccess;
import static net.openhft.chronicle.values.Values.newNativeReference;
public class PingPongLockLeft {
public static void main(String... ignored) throws IOException, InterruptedException {
ChronicleMap<String, BondVOInterface> chm = PingPongCASLeft.acquireCHM();
playPingPong(chm, 4, 5, true, "PingPongLockLEFT");
}
static void playPingPong(ChronicleMap<String, BondVOInterface> chm, double _coupon, double _coupon2, boolean setFirst, final String desc) throws InterruptedException {
BondVOInterface bond1 = newNativeReference(BondVOInterface.class);
BondVOInterface bond2 = newNativeReference(BondVOInterface.class);
BondVOInterface bond3 = newNativeReference(BondVOInterface.class);
BondVOInterface bond4 = newNativeReference(BondVOInterface.class);
chm.acquireUsing("369604101", bond1);
chm.acquireUsing("369604102", bond2);
chm.acquireUsing("369604103", bond3);
chm.acquireUsing("369604104", bond4);
System.out.printf("\n\n" + desc + ": Timing 1 x off-heap operations on " + chm.file() + "\n");
if (setFirst) {
bond1.setCoupon(_coupon);
bond2.setCoupon(_coupon);
bond3.setCoupon(_coupon);
bond4.setCoupon(_coupon);
}
int timeToCallNanoTime = 30;
int runs = 1000000;
long[] timings = new long[runs];
for (int j = 0; j < 10; j++) {
for (int i = 0; i < runs; i++) {
long _start = System.nanoTime();
toggleCoupon(bond1, _coupon, _coupon2);
toggleCoupon(bond2, _coupon, _coupon2);
toggleCoupon(bond3, _coupon, _coupon2);
toggleCoupon(bond4, _coupon, _coupon2);
timings[i] = (System.nanoTime() - _start - timeToCallNanoTime) / 4;
}
Arrays.sort(timings);
System.out.printf("#%d: lock,compare,set,unlock 50/90/99%%tile was %,d / %,d / %,d%n",
j, timings[runs / 2], timings[runs * 9 / 10], timings[runs * 99 / 100]);
}
}
private static void toggleCoupon(BondVOInterface bond, double _coupon, double _coupon2)
throws InterruptedException {
for (int i = 0; ; i++) {
AcquisitionStrategies.<ReadWriteLockingStrategy>spinLoopOrFail(5, TimeUnit.SECONDS)
.acquire(TryAcquireOperations.writeLock(),
VanillaReadWriteWithWaitsLockingStrategy.instance(),
checkedBytesStoreAccess(),
((Byteable) bond).bytesStore(), ((Byteable) bond).offset());
try {
if (bond.getCoupon() == _coupon) {
bond.setCoupon(_coupon2);
return;
}
if (i > 1000)
Thread.yield();
} finally {
VanillaReadWriteWithWaitsLockingStrategy.instance()
.writeUnlock(checkedBytesStoreAccess(),
((Byteable) bond).bytesStore(), ((Byteable) bond).offset());
}
}
}
// public static void main(String... ignored) throws IOException, InterruptedException {
// ChronicleMap<String, BondVOInterface> chm = PingPongCASLeft.acquireCHM();
//
// playPingPong(chm, 4, 5, true, "PingPongLockLEFT");
// }
//
// static void playPingPong(ChronicleMap<String, BondVOInterface> chm, double _coupon, double _coupon2, boolean setFirst, final String desc) throws InterruptedException {
// BondVOInterface bond1 = newNativeReference(BondVOInterface.class);
// BondVOInterface bond2 = newNativeReference(BondVOInterface.class);
// BondVOInterface bond3 = newNativeReference(BondVOInterface.class);
// BondVOInterface bond4 = newNativeReference(BondVOInterface.class);
//
// chm.acquireUsing("369604101", bond1);
// chm.acquireUsing("369604102", bond2);
// chm.acquireUsing("369604103", bond3);
// chm.acquireUsing("369604104", bond4);
// System.out.printf("\n\n" + desc + ": Timing 1 x off-heap operations on " + chm.file() + "\n");
// if (setFirst) {
// bond1.setCoupon(_coupon);
// bond2.setCoupon(_coupon);
// bond3.setCoupon(_coupon);
// bond4.setCoupon(_coupon);
// }
// int timeToCallNanoTime = 30;
// int runs = 1000000;
// long[] timings = new long[runs];
// for (int j = 0; j < 10; j++) {
// for (int i = 0; i < runs; i++) {
// long _start = System.nanoTime();
// toggleCoupon(bond1, _coupon, _coupon2);
// toggleCoupon(bond2, _coupon, _coupon2);
// toggleCoupon(bond3, _coupon, _coupon2);
// toggleCoupon(bond4, _coupon, _coupon2);
//
// timings[i] = (System.nanoTime() - _start - timeToCallNanoTime) / 4;
// }
// Arrays.sort(timings);
// System.out.printf("#%d: lock,compare,set,unlock 50/90/99%%tile was %,d / %,d / %,d%n",
// j, timings[runs / 2], timings[runs * 9 / 10], timings[runs * 99 / 100]);
// }
// }
//
// private static void toggleCoupon(BondVOInterface bond, double _coupon, double _coupon2)
// throws InterruptedException {
// for (int i = 0; ; i++) {
// AcquisitionStrategies.<ReadWriteLockingStrategy>spinLoopOrFail(5, TimeUnit.SECONDS)
// .acquire(TryAcquireOperations.writeLock(),
// VanillaReadWriteWithWaitsLockingStrategy.instance(),
// checkedBytesStoreAccess(),
// ((Byteable) bond).bytesStore(), ((Byteable) bond).offset());
// try {
// if (bond.getCoupon() == _coupon) {
// bond.setCoupon(_coupon2);
// return;
// }
// if (i > 1000)
// Thread.yield();
// } finally {
// VanillaReadWriteWithWaitsLockingStrategy.instance()
// .writeUnlock(checkedBytesStoreAccess(),
// ((Byteable) bond).bytesStore(), ((Byteable) bond).offset());
// }
// }
// }
}
src/test/java/net/openhft/chronicle/map/fromdocs/pingpong_latency/PingPongLockRight.java
+1 -1
View File
@@ -25,6 +25,6 @@ public class PingPongLockRight {
public static void main(String... ignored) throws IOException, InterruptedException {
ChronicleMap<String, BondVOInterface> chm = PingPongCASLeft.acquireCHM();
PingPongLockLeft.playPingPong(chm, 5, 4, false, "PingPongLockRIGHT");
//PingPongLockLeft.playPingPong(chm, 5, 4, false, "PingPongLockRIGHT");
}
}