pause support

Author: mnadeem

src/main/java/com/github/dexecutor/core/DefaultDexecutor.java

@@ -106,7 +106,7 @@ public void addAsDependencyToAllInitialNodes(final T nodeValue) {
 	}
 
 	@Override
-	public void recoverExecution(final ExecutionConfig config) {
+	public ExecutionResults<T, R> recoverExecution(final ExecutionConfig config) {
 		if (Phase.TERMINATED.equals(this.state.getCurrentPhase())) {
 			throw new IllegalStateException("Can't recover terminated dexecutor");		
 		} else {	
@@ -118,6 +118,7 @@ public void recoverExecution(final ExecutionConfig config) {
 			this.state.onTerminate();
 			logger.debug("Processed Nodes Ordering {}", this.state.getProcessedNodes());
 		}
+		return this.state.getExecutionResults();
 	}
 
 	public ExecutionResults<T, R> execute(final ExecutionConfig config) {
@@ -333,10 +334,11 @@ private void updateNode(final ExecutionResult<T, R> executionResult, final Node<
 
 	private void forceStopIfRequired() {
 		if (!shouldContinueProcessingNodes()) {
+			logger.debug("Force Stopping dexecutor");
 			this.state.forcedStop();
-			this.immediatelyRetryExecutor.shutdownNow();
-			this.scheduledRetryExecutor.shutdownNow();
-			this.timeoutExecutor.shutdownNow();
+			//this.immediatelyRetryExecutor.shutdownNow();
+			//this.scheduledRetryExecutor.shutdownNow();
+			//this.timeoutExecutor.shutdownNow();
 			throw new IllegalStateException("Forced to Stop the instance of Dexecutor!");
 		}		
 	}

src/main/java/com/github/dexecutor/core/Dexecutor.java

@@ -46,7 +46,7 @@ public interface Dexecutor<T, R> extends DependencyAware<T> {
 	 * After a dexecutor crash, create a new instance of dexecutor and call this method for recovery
 	 * @param config based on which execution would recover
 	 */
-	void recoverExecution(final ExecutionConfig config);
+	ExecutionResults<T, R> recoverExecution(final ExecutionConfig config);
 	/**
 	 * Prints the graph into the writer, using the traversar
 	 * 

src/test/java/com/github/dexecutor/core/DexecutorPauseAndRecover.java

@@ -0,0 +1,161 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 com.github.dexecutor.core;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.Test;
+
+import com.github.dexecutor.core.graph.Node;
+import com.github.dexecutor.core.support.TestUtil;
+import com.github.dexecutor.core.support.ThreadPoolUtil;
+import com.github.dexecutor.core.task.ExecutionResults;
+import com.github.dexecutor.core.task.Task;
+import com.github.dexecutor.core.task.TaskProvider;
+
+/**
+ * 
+ * @author Nadeem Mohammad
+ *
+ */
+public class DexecutorPauseAndRecover {
+	
+	@Test
+	public void testDependentTaskExecution() {
+		ExecutorService executorService = newExecutor();
+
+		try {
+			DexecutorConfig<Integer, Integer> config = new DexecutorConfig<>(executorService, new SleepyTaskProvider());
+
+			DefaultDexecutor<Integer, Integer> executor = new DefaultDexecutor<Integer, Integer>(config) {
+				Integer counter = 0;
+				
+				@Override
+				protected boolean shouldContinueProcessingNodes() {
+					counter++;
+					return counter != 2;
+				}
+			};
+
+			executor.addDependency(1, 2);
+			executor.addDependency(1, 2);
+			executor.addDependency(1, 3);
+			executor.addDependency(3, 4);
+			executor.addDependency(3, 5);
+			executor.addDependency(3, 6);
+			executor.addDependency(2, 7);
+			executor.addDependency(2, 9);
+			executor.addDependency(2, 8);
+			executor.addDependency(9, 10);
+			executor.addDependency(12, 13);
+			executor.addDependency(13, 4);
+			executor.addDependency(13, 14);
+			executor.addIndependent(11);
+
+			Thread thread = startRecoveryThread(executor);
+
+		    executeDexecutor(executor);
+			
+			waitForRecoveryThreadToFinish(thread);
+			
+
+
+			Collection<Node<Integer, Integer>> processedNodesOrder = TestUtil.processedNodesOrder(executor);
+			assertThat(processedNodesOrder).containsAll(executionOrderExpectedResult());
+			assertThat(processedNodesOrder).size().isGreaterThanOrEqualTo(4);
+			
+		} finally {
+			try {
+				executorService.shutdownNow();
+				executorService.awaitTermination(1, TimeUnit.SECONDS);
+			} catch (InterruptedException e) {
+				
+			}
+		}
+	}
+
+	private ExecutionResults<Integer, Integer> executeDexecutor(DefaultDexecutor<Integer, Integer> executor) {
+		ExecutionResults<Integer, Integer> result = null;
+		try {
+			result = executor.execute(ExecutionConfig.TERMINATING);
+		} catch (IllegalStateException e) {
+			e.printStackTrace();
+		}
+		return result;
+	}
+
+	private void waitForRecoveryThreadToFinish(Thread thread) {
+		try {
+			thread.join();
+		} catch (InterruptedException e) {
+			e.printStackTrace();
+		}
+	}
+
+	private Thread startRecoveryThread(DefaultDexecutor<Integer, Integer> executor) {
+		Runnable task = () -> {
+		    
+		    try {
+				TimeUnit.SECONDS.sleep(1);
+				ExecutionResults<Integer, Integer> results = executor.recoverExecution(ExecutionConfig.TERMINATING);
+				System.out.println(results);
+			} catch (InterruptedException e) {
+				e.printStackTrace();
+			}
+		    
+		};
+
+		Thread thread = new Thread(task);
+		thread.start();
+		return thread;
+	}
+
+	private Collection<Node<Integer, Integer>> executionOrderExpectedResult() {
+		List<Node<Integer, Integer>> result = new ArrayList<Node<Integer, Integer>>();
+		result.add(new Node<Integer, Integer>(1));
+		result.add(new Node<Integer, Integer>(2));
+		result.add(new Node<Integer, Integer>(11));
+		result.add(new Node<Integer, Integer>(12));
+		return result;
+	}
+	private ExecutorService newExecutor() {
+		return Executors.newFixedThreadPool(ThreadPoolUtil.ioIntesivePoolSize());
+	}
+
+	private static class SleepyTaskProvider implements TaskProvider<Integer, Integer> {
+		
+		public Task<Integer, Integer> provideTask(final Integer id) {
+
+			return new Task<Integer, Integer>() {
+
+				private static final long serialVersionUID = 1L;
+
+				public Integer execute() {				
+					return id;
+				}
+			};			
+		}		
+	}
+}