v2.0.2: upgraded the async runnable implementation

Author: Andrey Kurilov

build.gradle

@@ -11,7 +11,7 @@ allprojects {
 	apply plugin: "maven"
 	apply plugin: "signing"
 	group = "com.github.akurilov"
-	version = "2.0.1"
+	version = "2.0.2"
 }
 
 ext.moduleName = "${group}.concurrent"
@@ -21,7 +21,7 @@ repositories {
 }
 
 dependencies {
-	compile("com.github.akurilov:java-commons:[2.0.1,)")
+	compile("com.github.akurilov:java-commons:[2.0.3,)")
 	testCompile("junit:junit:4.12")
 }
 

src/main/java/com/github/akurilov/concurrent/AsyncRunnableBase.java

@@ -1,9 +1,10 @@
 package com.github.akurilov.concurrent;
 
 import java.io.IOException;
-import java.rmi.RemoteException;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
 
 import static com.github.akurilov.concurrent.AsyncRunnable.State.FINISHED;
 import static com.github.akurilov.concurrent.AsyncRunnable.State.INITIAL;
@@ -14,117 +15,156 @@
 public abstract class AsyncRunnableBase
 implements AsyncRunnable {
 
-	private final AtomicReference<State> stateRef = new AtomicReference<>(INITIAL);
-	protected final Object state = new Object();
+	private volatile State state = State.INITIAL;
+	private final Lock stateLock = new ReentrantLock();
+	protected final Condition stateChanged = stateLock.newCondition();
 
 	@Override
 	public final State state() {
-		return stateRef.get();
+		return state;
 	}
 
 	@Override
 	public boolean isInitial() {
-		return INITIAL.equals(stateRef.get());
+		return INITIAL == state;
 	}
 
 	@Override
 	public boolean isStarted() {
-		return STARTED.equals(stateRef.get());
+		return STARTED == state;
 	}
 
 	@Override
 	public boolean isShutdown() {
-		return SHUTDOWN.equals(stateRef.get());
+		return SHUTDOWN == state;
 	}
 
 	@Override
 	public boolean isStopped() {
-		return STOPPED.equals(stateRef.get());
+		return STOPPED == state;
 	}
 
 	@Override
 	public boolean isFinished() {
-		return FINISHED.equals(stateRef.get());
+		return FINISHED == state;
 	}
 
 	@Override
 	public boolean isClosed() {
-		return null == stateRef.get();
+		return null == state;
 	}
 
 	@Override
 	public final AsyncRunnableBase start()
 	throws IllegalStateException {
-		if(stateRef.compareAndSet(INITIAL, STARTED) || stateRef.compareAndSet(STOPPED, STARTED)) {
-			synchronized(state) {
-				doStart();
-				state.notifyAll();
+		if(stateLock.tryLock()) {
+			try {
+				if(state == INITIAL || state == STOPPED) {
+					doStart();
+					state = STARTED;
+					stateChanged.signalAll();
+				} else {
+					throw new IllegalStateException(
+						"Not allowed to start while state is \"" + state + "\""
+					);
+				}
+			} finally {
+				stateLock.unlock();
 			}
 		} else {
-			throw new IllegalStateException(
-				"Not allowed to start while state is \"" + stateRef.get() + "\""
-			);
+			throw new IllegalStateException("Start: failed to acquire the state lock");
 		}
 		return this;
 	}
 
 	@Override
 	public final AsyncRunnableBase shutdown()
 	throws IllegalStateException {
-		if(stateRef.compareAndSet(STARTED, SHUTDOWN)) {
-			synchronized(state) {
-				doShutdown();
-				state.notifyAll();
+		if(stateLock.tryLock()) {
+			try {
+				if(state == STARTED) {
+					doShutdown();
+					state = SHUTDOWN;
+					stateChanged.signalAll();
+				} else {
+					throw new IllegalStateException(
+						"Not allowed to shutdown while state is \"" + state + "\""
+					);
+				}
+			} finally {
+				stateLock.unlock();
 			}
 		} else {
-			throw new IllegalStateException(
-				"Not allowed to shutdown while state is \"" + stateRef.get() + "\""
-			);
+			throw new IllegalStateException("Shutdown: failed to acquire the state lock");
 		}
 		return this;
 	}
 
 	@Override
 	public final AsyncRunnableBase stop()
-	throws IllegalStateException, RemoteException {
-		try {
-			shutdown();
-		} catch(final IllegalStateException ignored) {
-		}
-		if(stateRef.compareAndSet(STARTED, STOPPED) || stateRef.compareAndSet(SHUTDOWN, STOPPED)) {
-			synchronized(state) {
-				doStop();
-				state.notifyAll();
+	throws IllegalStateException {
+		if(stateLock.tryLock()) {
+			try {
+				if(state == STARTED || state == SHUTDOWN) {
+					doStop();
+					state = STOPPED;
+					stateChanged.signalAll();
+				} else {
+					throw new IllegalStateException(
+						"Not allowed to stop while state is \"" + state + "\""
+					);
+				}
+			} finally {
+				stateLock.unlock();
 			}
 		} else {
-			throw new IllegalStateException(
-				"Not allowed to stop while state is \"" + stateRef.get() + "\""
-			);
+			throw new IllegalStateException("Stop: failed to acquire the state lock");
 		}
 		return this;
 	}
 
 	@Override
 	public final AsyncRunnableBase await()
 	throws IllegalStateException, InterruptedException {
-		await(Long.MAX_VALUE, TimeUnit.DAYS);
+		await(Long.MAX_VALUE, TimeUnit.MILLISECONDS);
 		return this;
 	}
 
 	@Override
 	public boolean await(final long timeout, final TimeUnit timeUnit)
 	throws IllegalStateException, InterruptedException {
-		final long timeOutMilliSec = timeUnit.toMillis(timeout);
-		final long t = System.currentTimeMillis();
-		while(isStarted() || isShutdown()) {
-			if(System.currentTimeMillis() - t >= timeOutMilliSec) {
-				return false;
-			}
-			synchronized(state) {
-				state.wait(100);
+		final long invokeTimeMillis = System.currentTimeMillis();
+		final long timeOutMillis = timeUnit.toMillis(timeout);
+		long elapsedTimeMillis;
+		while(timeOutMillis > (elapsedTimeMillis = System.currentTimeMillis() - invokeTimeMillis)) {
+			if(state != STARTED && state != SHUTDOWN) {
+				return true; // condition is reached
+			} else {
+				if(stateLock.tryLock(timeOutMillis - elapsedTimeMillis, TimeUnit.MILLISECONDS)) {
+					try {
+						// spent a time to wait for the state lock, need to update the elapsed time
+						elapsedTimeMillis = System.currentTimeMillis() - invokeTimeMillis;
+						// recheck for the timeout condition
+						if(timeOutMillis > elapsedTimeMillis) {
+							if(
+								stateChanged.await(
+									timeOutMillis - elapsedTimeMillis, TimeUnit.MILLISECONDS
+								)
+							) { // the state is changed, recheck the condition
+								if(state != STARTED && state != SHUTDOWN) {
+									return true;
+								} // continue otherwise (no timeout yet, condition is not reached)
+							}
+						} else { // timeout, exit the loop
+							break;
+						}
+					} finally {
+						stateLock.unlock();
+					}
+				}
 			}
 		}
-		return true;
+		return state != STARTED && state != SHUTDOWN;
 	}
 
 	@Override
@@ -136,12 +176,18 @@ public void close()
 		} catch(final IllegalStateException ignored) {
 		}
 		// then close actually
-		synchronized(state) {
-			if(null != stateRef.get()) {
-				doClose();
-				stateRef.set(null);
-				state.notifyAll();
+		if(stateLock.tryLock()) {
+			try {
+				if(null != state) {
+					doClose();
+					state = null;
+					stateChanged.signalAll();
+				}
+			} finally {
+				stateLock.unlock();
 			}
+		} else {
+			throw new IllegalStateException("Close: failed to acquire the state lock");
 		}
 	}
 

src/main/java/com/github/akurilov/concurrent/coroutines/Coroutine.java renamed to src/main/java/com/github/akurilov/concurrent/coroutine/Coroutine.java

@@ -1,4 +1,4 @@
-package com.github.akurilov.concurrent.coroutines;
+package com.github.akurilov.concurrent.coroutine;
 
 import com.github.akurilov.concurrent.AsyncRunnable;
 

src/main/java/com/github/akurilov/concurrent/coroutines/CoroutineBase.java renamed to src/main/java/com/github/akurilov/concurrent/coroutine/CoroutineBase.java

@@ -1,18 +1,14 @@
-package com.github.akurilov.concurrent.coroutines;
+package com.github.akurilov.concurrent.coroutine;
 
 import com.github.akurilov.concurrent.AsyncRunnableBase;
 
-import java.util.logging.Logger;
-
 /**
  * The base class for all coroutines.
  */
 public abstract class CoroutineBase
 extends AsyncRunnableBase
 implements Coroutine {
 
-	private static final Logger LOG = Logger.getLogger(CoroutineBase.class.getName());
-
 	private final CoroutinesExecutor executor;
 
 	protected CoroutineBase(final CoroutinesExecutor executor) {

src/main/java/com/github/akurilov/concurrent/coroutines/CoroutinesExecutor.java renamed to src/main/java/com/github/akurilov/concurrent/coroutine/CoroutinesExecutor.java

@@ -1,4 +1,4 @@
-package com.github.akurilov.concurrent.coroutines;
+package com.github.akurilov.concurrent.coroutine;
 
 import com.github.akurilov.concurrent.ContextAwareThreadFactory;
 

src/main/java/com/github/akurilov/concurrent/coroutines/CoroutinesExecutorTask.java renamed to src/main/java/com/github/akurilov/concurrent/coroutine/CoroutinesExecutorTask.java

@@ -1,4 +1,4 @@
-package com.github.akurilov.concurrent.coroutines;
+package com.github.akurilov.concurrent.coroutine;
 
 import com.github.akurilov.concurrent.AsyncRunnableBase;
 

src/main/java/com/github/akurilov/concurrent/coroutines/ExclusiveCoroutineBase.java renamed to src/main/java/com/github/akurilov/concurrent/coroutine/ExclusiveCoroutineBase.java

@@ -1,4 +1,4 @@
-package com.github.akurilov.concurrent.coroutines;
+package com.github.akurilov.concurrent.coroutine;
 
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;

src/main/java/com/github/akurilov/concurrent/coroutines/OutputCoroutine.java renamed to src/main/java/com/github/akurilov/concurrent/coroutine/OutputCoroutine.java

@@ -1,4 +1,4 @@
-package com.github.akurilov.concurrent.coroutines;
+package com.github.akurilov.concurrent.coroutine;
 
 import com.github.akurilov.commons.io.Output;
 

src/main/java/com/github/akurilov/concurrent/coroutines/RoundRobinOutputCoroutine.java renamed to src/main/java/com/github/akurilov/concurrent/coroutine/RoundRobinOutputCoroutine.java

@@ -1,4 +1,4 @@
-package com.github.akurilov.concurrent.coroutines;
+package com.github.akurilov.concurrent.coroutine;
 
 import com.github.akurilov.commons.collection.OptLockArrayBuffer;
 import com.github.akurilov.commons.collection.OptLockBuffer;

src/main/java/com/github/akurilov/concurrent/coroutines/TransferCoroutine.java renamed to src/main/java/com/github/akurilov/concurrent/coroutine/TransferCoroutine.java

@@ -1,4 +1,4 @@
-package com.github.akurilov.concurrent.coroutines;
+package com.github.akurilov.concurrent.coroutine;
 
 import com.github.akurilov.commons.collection.OptLockArrayBuffer;
 import com.github.akurilov.commons.collection.OptLockBuffer;

src/main/java/com/github/akurilov/concurrent/coroutines/example/HelloWorldCoroutine.java renamed to src/main/java/com/github/akurilov/concurrent/coroutine/example/HelloWorldCoroutine.java

@@ -1,7 +1,7 @@
-package com.github.akurilov.concurrent.coroutines.example;
+package com.github.akurilov.concurrent.coroutine.example;
 
-import com.github.akurilov.concurrent.coroutines.CoroutineBase;
-import com.github.akurilov.concurrent.coroutines.CoroutinesExecutor;
+import com.github.akurilov.concurrent.coroutine.CoroutineBase;
+import com.github.akurilov.concurrent.coroutine.CoroutinesExecutor;
 
 import java.io.IOException;
 

src/main/java/com/github/akurilov/concurrent/coroutines/example/HelloWorldExclusiveCoroutine.java renamed to src/main/java/com/github/akurilov/concurrent/coroutine/example/HelloWorldExclusiveCoroutine.java

@@ -1,7 +1,7 @@
-package com.github.akurilov.concurrent.coroutines.example;
+package com.github.akurilov.concurrent.coroutine.example;
 
-import com.github.akurilov.concurrent.coroutines.CoroutinesExecutor;
-import com.github.akurilov.concurrent.coroutines.ExclusiveCoroutineBase;
+import com.github.akurilov.concurrent.coroutine.CoroutinesExecutor;
+import com.github.akurilov.concurrent.coroutine.ExclusiveCoroutineBase;
 
 import java.io.IOException;
 

src/main/java/com/github/akurilov/concurrent/coroutines/example/Main.java renamed to src/main/java/com/github/akurilov/concurrent/coroutine/example/Main.java

@@ -1,7 +1,7 @@
-package com.github.akurilov.concurrent.coroutines.example;
+package com.github.akurilov.concurrent.coroutine.example;
 
-import com.github.akurilov.concurrent.coroutines.Coroutine;
-import com.github.akurilov.concurrent.coroutines.CoroutinesExecutor;
+import com.github.akurilov.concurrent.coroutine.Coroutine;
+import com.github.akurilov.concurrent.coroutine.CoroutinesExecutor;
 
 import java.io.IOException;
 import java.util.concurrent.TimeUnit;

src/main/java/com/github/akurilov/concurrent/IndexThrottle.java renamed to src/main/java/com/github/akurilov/concurrent/throttle/IndexThrottle.java

@@ -1,4 +1,4 @@
-package com.github.akurilov.concurrent;
+package com.github.akurilov.concurrent.throttle;
 
 public interface IndexThrottle {
 

src/main/java/com/github/akurilov/concurrent/RateThrottle.java renamed to src/main/java/com/github/akurilov/concurrent/throttle/RateThrottle.java

@@ -1,4 +1,4 @@
-package com.github.akurilov.concurrent;
+package com.github.akurilov.concurrent.throttle;
 
 import java.util.concurrent.TimeUnit;
 

src/main/java/com/github/akurilov/concurrent/SequentialWeightsThrottle.java renamed to src/main/java/com/github/akurilov/concurrent/throttle/SequentialWeightsThrottle.java

@@ -1,4 +1,4 @@
-package com.github.akurilov.concurrent;
+package com.github.akurilov.concurrent.throttle;
 
 import java.util.Arrays;
 

src/main/java/com/github/akurilov/concurrent/Throttle.java renamed to src/main/java/com/github/akurilov/concurrent/throttle/Throttle.java

@@ -1,4 +1,4 @@
-package com.github.akurilov.concurrent;
+package com.github.akurilov.concurrent.throttle;
 
 /**
  Throttle can make a decision about the specified thing to pass or to wait.