mirror of
https://github.com/OpenHFT/Chronicle-Map.git
synced 2021-09-13 01:26:12 +00:00
using chm as StampedLock readerCount,lockState, lastWriter custodian
This commit is contained in:
+111
-9
@@ -9,10 +9,13 @@ 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 {
|
||||
|
||||
@@ -21,18 +24,22 @@ public class ChronicleStampedLock extends StampedLock {
|
||||
newNativeReference(ChronicleStampedLockVOInterface.class);
|
||||
ChronicleStampedLockVOInterface lastWriterT =
|
||||
newNativeReference(ChronicleStampedLockVOInterface.class);
|
||||
|
||||
ChronicleStampedLockVOInterface readerCount =
|
||||
newNativeReference(ChronicleStampedLockVOInterface.class);
|
||||
|
||||
ChronicleStampedLock(String chronicelStampedLockLocality) {
|
||||
try {
|
||||
chm = DirtyReadTolerance.offHeapLock(chronicelStampedLockLocality);
|
||||
chm = offHeapLock(chronicelStampedLockLocality);
|
||||
chm.acquireUsing("Stamp ", offHeapLock);
|
||||
chm.acquireUsing("LastWriterTime ", lastWriterT);
|
||||
chm.acquireUsing("ReaderCount ", readerCount);
|
||||
System.out.println(
|
||||
" ,@t=" + System.currentTimeMillis() +
|
||||
" ChronicleStampedLock constructed" +
|
||||
","
|
||||
);
|
||||
readerCount.setEntryLockState(0L);
|
||||
chm.put("ReadederCount ", readerCount);
|
||||
} catch (IOException e) {
|
||||
e.printStackTrace();
|
||||
}
|
||||
@@ -91,10 +98,30 @@ public class ChronicleStampedLock extends StampedLock {
|
||||
|
||||
offHeapLock = chm.get("Stamp ");
|
||||
lastWriterT = chm.get("LastWriterTime ");
|
||||
do {
|
||||
l = offHeapLock.getEntryLockState();
|
||||
} while (l != 0L);
|
||||
readerCount = chm.get("ReaderCount ");
|
||||
|
||||
do {
|
||||
System.out.println(
|
||||
" ,@t=" + System.currentTimeMillis() +
|
||||
" ChronicleStampedLock tryWriterLock() WAITING " +
|
||||
" on offHeapLock.unlock(" +
|
||||
offHeapLock.getEntryLockState() +
|
||||
") ,"
|
||||
);
|
||||
offHeapLock = chm.get("Stamp ");
|
||||
l = offHeapLock.getEntryLockState();
|
||||
readerCount = chm.get("ReaderCount ");
|
||||
try {
|
||||
Thread.sleep(1000);
|
||||
} catch (InterruptedException e) {
|
||||
e.printStackTrace();
|
||||
}
|
||||
} while (l != 0L || readerCount.getEntryLockState() > 0);
|
||||
System.out.println(
|
||||
" ,@t=" + System.currentTimeMillis() +
|
||||
" ChronicleStampedLock tryWriterLock() RESUMING " +
|
||||
" ,"
|
||||
);
|
||||
long t = System.currentTimeMillis();
|
||||
offHeapLock.setEntryLockState(-t); //negative ==> Writer holds StampedLock
|
||||
lastWriterT.setEntryLockState(t);
|
||||
@@ -109,6 +136,84 @@ public class ChronicleStampedLock extends StampedLock {
|
||||
return (offHeapLock.getEntryLockState());
|
||||
}
|
||||
|
||||
@Override
|
||||
public long tryReadLock() {
|
||||
long l = 0;
|
||||
|
||||
offHeapLock = chm.get("Stamp ");
|
||||
lastWriterT = chm.get("LastWriterTime ");
|
||||
readerCount = chm.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);
|
||||
System.out.println(
|
||||
" ,@t=" + System.currentTimeMillis() +
|
||||
" ChronicleStampedLock tryReadLock() RESUMING ,"
|
||||
);
|
||||
|
||||
readerCount.addAtomicReaderCount( 1);
|
||||
|
||||
long t = System.currentTimeMillis();
|
||||
offHeapLock.setEntryLockState(t); //negative ==> Writer holds StampedLock
|
||||
|
||||
chm.put("Stamp ", offHeapLock);
|
||||
chm.put("ReaderCount ", readerCount);
|
||||
|
||||
System.out.println(
|
||||
" ,@t=" + t +
|
||||
" ChronicleStampedLock tryReadLock() returned stamp=" +
|
||||
offHeapLock.getEntryLockState() +
|
||||
" readerCount=["+readerCount.getReaderCount()+"]"+
|
||||
","
|
||||
);
|
||||
return (offHeapLock.getEntryLockState());
|
||||
}
|
||||
|
||||
@Override
|
||||
public void unlockRead(long stamp) {
|
||||
offHeapLock = chm.get("Stamp ");
|
||||
readerCount = chm.get("ReaderCount ");
|
||||
System.out.println(
|
||||
" ,@t=" + System.currentTimeMillis() +
|
||||
" ChronicleStampedLock unlockRead(" + stamp + ") unlocking.."+
|
||||
"ReaderCount="+readerCount.getReaderCount()+
|
||||
","
|
||||
);
|
||||
|
||||
readerCount.addAtomicReaderCount(-1);
|
||||
if (readerCount.getReaderCount() == 0) {
|
||||
offHeapLock.setEntryLockState(0L);
|
||||
}
|
||||
System.out.println(
|
||||
" ,@t=" + System.currentTimeMillis() +
|
||||
"ReaderCount="+readerCount.getReaderCount()+
|
||||
","
|
||||
);
|
||||
chm.put("Stamp ", offHeapLock);
|
||||
chm.put("ReaderCount ", readerCount);
|
||||
System.out.println(
|
||||
" ,@t=" + System.currentTimeMillis() +
|
||||
" ChronicleStampedLock unlockRead(" +
|
||||
stamp +
|
||||
") unlocked. set to Zero. ReaderCount=" +
|
||||
readerCount.getReaderCount() +
|
||||
","
|
||||
);
|
||||
}
|
||||
|
||||
|
||||
@Override
|
||||
public void unlockWrite(long stamp) {
|
||||
System.out.println(
|
||||
@@ -129,7 +234,7 @@ public class ChronicleStampedLock extends StampedLock {
|
||||
|
||||
static ChronicleMap<String, ChronicleStampedLockVOInterface> offHeapLock(String operand)
|
||||
throws IOException {
|
||||
// ensure thread ids are globally unique.
|
||||
|
||||
AffinitySupport.setThreadId();
|
||||
return ChronicleMapBuilder.of(String.class, ChronicleStampedLockVOInterface.class)
|
||||
.entries(16)
|
||||
@@ -137,10 +242,7 @@ public class ChronicleStampedLock extends StampedLock {
|
||||
.createPersistedTo(
|
||||
new File(
|
||||
operand
|
||||
// "C:\\Users\\buddy\\dev\\shm\\OPERAND_CHRONICLE_MAP"
|
||||
)
|
||||
);
|
||||
//.create();
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
+7
@@ -27,5 +27,12 @@ interface ChronicleStampedLockVOInterface {
|
||||
|
||||
void setEntryLockState(long entryLockState);
|
||||
|
||||
@Group(1)
|
||||
long getReaderCount();
|
||||
|
||||
void setReaderCount(long rc); /* time in millis */
|
||||
|
||||
long addAtomicReaderCount(long toAdd);
|
||||
|
||||
|
||||
}
|
||||
|
||||
+101
@@ -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();
|
||||
}
|
||||
}
|
||||
}
|
||||
+10
-6
@@ -1,7 +1,7 @@
|
||||
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.BondVOInterface;;
|
||||
|
||||
import java.util.Scanner;
|
||||
import java.util.concurrent.locks.StampedLock;
|
||||
@@ -16,10 +16,11 @@ public class DirtyReadOffender {
|
||||
try {
|
||||
String isoLevel = args[0];
|
||||
long sleepT = Long.parseLong(args[1]);
|
||||
long holdTime = Long.parseLong(args[2]);
|
||||
|
||||
ChronicleMap<String, BondVOInterface> chm =
|
||||
DirtyReadTolerance.offHeap(
|
||||
args[2]
|
||||
args[3]
|
||||
+ "OPERAND_CHRONICLE_MAP"
|
||||
);
|
||||
System.out.println(
|
||||
@@ -27,7 +28,7 @@ public class DirtyReadOffender {
|
||||
" DirtyReadOffender established chm "
|
||||
);
|
||||
StampedLock offHeapLock = new ChronicleStampedLock(
|
||||
args[2]
|
||||
args[3]
|
||||
+ "OPERAND_ChronicleStampedLock"
|
||||
);
|
||||
BondVOInterface bond = newNativeReference(BondVOInterface.class);
|
||||
@@ -55,7 +56,6 @@ public class DirtyReadOffender {
|
||||
" @t=" + System.currentTimeMillis() +
|
||||
" DirtyReadOffender ACQUIRING offHeapLock.writeLock();"
|
||||
);
|
||||
|
||||
while ((stamp = offHeapLock.tryWriteLock()) == 0) {
|
||||
;
|
||||
}
|
||||
@@ -81,6 +81,11 @@ public class DirtyReadOffender {
|
||||
"] 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() +
|
||||
@@ -103,5 +108,4 @@ public class DirtyReadOffender {
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
+1
-17
@@ -24,8 +24,6 @@ public class DirtyReadTolerance<K,V> {
|
||||
+ "OPERAND_CHRONICLE_MAP"
|
||||
);
|
||||
System.out.println(",Established chm,");
|
||||
|
||||
|
||||
}
|
||||
|
||||
static ChronicleMap<String, BondVOInterface> offHeap(String operand) throws IOException {
|
||||
@@ -43,19 +41,5 @@ public class DirtyReadTolerance<K,V> {
|
||||
//.create();
|
||||
}
|
||||
|
||||
static ChronicleMap<String, ChronicleStampedLockVOInterface> offHeapLock(String operand)
|
||||
throws IOException {
|
||||
// ensure thread ids are globally unique.
|
||||
AffinitySupport.setThreadId();
|
||||
return ChronicleMapBuilder.of(String.class, ChronicleStampedLockVOInterface.class)
|
||||
.entries(16)
|
||||
.averageKeySize("123456789".length())
|
||||
.createPersistedTo(
|
||||
new File(
|
||||
operand
|
||||
// "C:\\Users\\buddy\\dev\\shm\\OPERAND_CHRONICLE_MAP"
|
||||
)
|
||||
);
|
||||
//.create();
|
||||
}
|
||||
|
||||
}
|
||||
+11
-4
@@ -14,12 +14,13 @@ public class DirtyReadVictim {
|
||||
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[2]
|
||||
args[3]
|
||||
+ "OPERAND_CHRONICLE_MAP"
|
||||
);
|
||||
Double coupon = 0.00;
|
||||
@@ -32,9 +33,15 @@ public class DirtyReadVictim {
|
||||
);
|
||||
StampedLock offHeapLock =
|
||||
new ChronicleStampedLock(
|
||||
args[2]
|
||||
args[3]
|
||||
+ "OPERAND_ChronicleStampedLock"
|
||||
);
|
||||
System.out.println(
|
||||
" ,,@t=" + System.currentTimeMillis() +
|
||||
" DirtyReadVictim sleeping " + sleepMock + " seconds"
|
||||
);
|
||||
|
||||
Thread.sleep(sleepMock * 1_000);
|
||||
while ((stamp = offHeapLock.tryOptimisticRead()) < 0) {
|
||||
;
|
||||
}
|
||||
@@ -60,10 +67,10 @@ public class DirtyReadVictim {
|
||||
);
|
||||
System.out.println(
|
||||
" ,,@t=" + System.currentTimeMillis() +
|
||||
" DirtyReadVictim sleeping " + sleepMock + " seconds"
|
||||
" DirtyReadVictim sleeping " + holdTime + " seconds"
|
||||
);
|
||||
|
||||
Thread.sleep(sleepMock * 1_000);
|
||||
Thread.sleep(holdTime * 1_000);
|
||||
System.out.println(
|
||||
" ,,@t=" + System.currentTimeMillis() +
|
||||
" DirtyReadVictim awakening"
|
||||
|
||||
BIN
Binary file not shown.
Binary file not shown.
Reference in New Issue
Block a user