非阻塞同步算法实战（三）－LatestResultsProvider

/**
* @author trytocatch@163.com
* @date 2013-2-2
*/
public abstract class LatestResultsProvider {
/** update return value */
public static final int UPDATE_FAILED = -1;
public static final int UPDATE_NO_NEED_TO_UPDATE = 0;
public static final int UPDATE_SUCCESS = 1;
public static final int UPDATE_COMMITTED = 2;
/** update return value */ /** work states*/
private static final int WS_OFF = 0;
private static final int WS_NEW_TASK = 1;
private static final int WS_WORKING = 2;
private static final int WS_DELAYING = 3;
private static final int WS_DELAY_RESET = 4;
private static final int WS_CANCELED = 5;
/** work states*/
private final AtomicInteger workState; private int sleepPeriod = 30; private final AtomicInteger parametersVersion;
private volatile int updateDelay;// updateDelay>=0
private volatile int delayUpperLimit; private final BoundlessCyclicBarrier barrier;
private Thread workThread; /**
*
* @param updateDelay unit: millisecond
* @param delayUpperLimit limit the sum of the delay, disabled
* while delayUpperLimit<0, unit: millisecond
*/
public LatestResultsProvider(int updateDelay, int delayUpperLimit) {
if (updateDelay < 0)
this.updateDelay = 0;
else
this.updateDelay = updateDelay;
this.delayUpperLimit = delayUpperLimit;
barrier = new BoundlessCyclicBarrier(0);
workState = new AtomicInteger(WS_OFF);
parametersVersion = new AtomicInteger(0);
initThread();
} private void initThread() {
workThread = new Thread("trytocatch's worker") {
@Override
public void run() {
int sleepCount = 0;
for (;;) {
try {
while (!workState.compareAndSet(WS_NEW_TASK,
updateDelay > 0 ? WS_DELAY_RESET : WS_WORKING)) {
if (workState.compareAndSet(WS_CANCELED, WS_OFF)) {
barrier.cancel();
}
LockSupport.park();
interrupted();
}
if (workState.get() == WS_DELAY_RESET) {
int delaySum = 0;
for (;;) {
if (workState.compareAndSet(WS_DELAY_RESET,
WS_DELAYING)) {
sleepCount = (updateDelay + sleepPeriod - 1)
/ sleepPeriod;
}
sleep(sleepPeriod);
if (--sleepCount <= 0
&& workState.compareAndSet(WS_DELAYING,
WS_WORKING))
break;
if (delayUpperLimit >= 0) {
delaySum += sleepPeriod;
if (delaySum >= delayUpperLimit) {
if (!workState.compareAndSet(
WS_DELAYING, WS_WORKING))
workState.compareAndSet(
WS_DELAY_RESET, WS_WORKING);
break;
}
}
if (workState.get() != WS_DELAYING
&& workState.get() != WS_DELAY_RESET)
break;
}
}
if (isWorking()) {
int workingVersion = parametersVersion.get();
try {
calculateResult();
if (workState.compareAndSet(WS_WORKING, WS_OFF))
barrier.nextCycle(workingVersion);
} catch (Throwable t) {
t.printStackTrace();
workState.set(WS_CANCELED);
}
}
} catch (InterruptedException e) {
workState.compareAndSet(WS_DELAYING, WS_CANCELED);
workState.compareAndSet(WS_DELAY_RESET, WS_CANCELED);
}
}// for(;;)
}// run()
};
workThread.setDaemon(true);
workThread.start();
} public int getUpdateDelay() {
return updateDelay;
} /**
* @param updateDelay
* delay time. unit: millisecond
*/
public void setUpdateDelay(int updateDelay) {
this.updateDelay = updateDelay < 0 ? 0 : updateDelay;
} public int getDelayUpperLimit() {
return delayUpperLimit;
} /**
* @param delayUpperLimit limit the sum of the delay, disabled
* while delayUpperLimit<0, unit: millisecond
*/
public void setDelayUpperLimit(int delayUpperLimit) {
this.delayUpperLimit = delayUpperLimit;
} public final void stopCurrentWorking() {
workState.set(WS_CANCELED);
} /**
* @return NO_NEED_TO_UPDATE, COMMITTED
*/
public final int update() {
if (isResultUptodate())
return UPDATE_NO_NEED_TO_UPDATE;
if (workState.compareAndSet(WS_CANCELED, WS_NEW_TASK)
|| workState.compareAndSet(WS_OFF, WS_NEW_TASK))
LockSupport.unpark(workThread);
return UPDATE_COMMITTED;
} /**
* @param timeout
* unit:nanoseconds
* @return FAILED, NO_NEED_TO_UPDATE, SUCCESS
* @throws InterruptedException
*/
public final int updateAndWait(long nanosTimeout)
throws InterruptedException {
int newVersion = parametersVersion.get();
if (update() == UPDATE_NO_NEED_TO_UPDATE)
return UPDATE_NO_NEED_TO_UPDATE;
barrier.awaitWithAssignedVersion(newVersion, nanosTimeout);
return barrier.getVersion() - newVersion >= 0 ? UPDATE_SUCCESS
: UPDATE_FAILED;
} /**
* @return FAILED, NO_NEED_TO_UPDATE, SUCCESS
* @throws InterruptedException
*/
public final int updateAndWait() throws InterruptedException {
return updateAndWait(0);
} public final boolean isResultUptodate() {
return parametersVersion.get() == barrier.getVersion();
} /**
* be used in calculateResult()
* @return true: the work state is working, worth to calculate the
* result absolutely, otherwise you can cancel the current calculation
*/
protected final boolean isWorking() {
return workState.get()==WS_WORKING;
} /**
* you must call this after update the parameters, and before calling the
* update
*/
protected final void updateParametersVersion() {
int pVersion = parametersVersion.get();
//CAS failed means that another thread do the same work already
if (parametersVersion.compareAndSet(pVersion, pVersion + 1))
if (!workState.compareAndSet(WS_DELAYING, WS_DELAY_RESET))
workState.compareAndSet(WS_WORKING, WS_NEW_TASK);
} /**
* implement this to deal with you task
*/
protected abstract void calculateResult();
}