package com.stuelken.java.b4.parallel.c09.concurrentstealing;

import java.util.concurrent.*;

/**
 * Es soll folgende Tasks geben: A ist der erste Task. Daraus sollen 3 Tasks B1,
 * B2, B3 als "Fork" erstellt werden. Abschließend sollen die drei wieder als
 * "Join" zu C fortgesetzt werden.
 * 
 * @author t2m
 */
public class WorkStealingForkJoinDemo {

	/**
	 * Oberaufgabe oder erster Task, der von drei Unteraufgaben B1–B3 "forkt" und
	 * deren Ergebnisse "joint".
	 * 
	 * Erweitert die Klasse {@link RecursiveTask} mit dem
	 * Typparameter {@link Integer}, damit unserer Task
	 * schlussendlich auch ein Ergebnis berechnen kann. 
	 * 
	 * @author t2m
	 */
	static class TaskA extends RecursiveTask<Integer> {

		/**
		 * 
		 */
		private static final long serialVersionUID = -506106973422841988L;

		/**
		 * "The main computation performed by this task." (JavaDoc).
		 */
		@Override
		protected Integer compute() {
			
			System.out.println("TaskA compute begonnen.");

			// Wir erzeugen 3 Tasks für unseren Fork, welche aber
			// alle auf dem selben Typ basieren und sich nur über den
			// Wert für den Konstruktor unterscheiden.
			
			System.out.println("TaskA erzeugt b1, 2, b3 mit b1.fork() etc..");
			
			TaskB b1 = new TaskB(1); // Aufgabe B mit Wert 1
			TaskB b2 = new TaskB(2); // Aufgabe B mit Wert 2
			TaskB b3 = new TaskB(3); // Aufgabe B mit Wert 3

			// Fork der Unteraufgaben
			// Dieser Befehl bewirkt, dass sich dieser Task in dem Container-Thread,
			// in welchem er sich befindet, in einen "common pool" einzufinden
			// versucht, um dort asynchron nebenläufig mit anderen weiteren
			// Tasks ausgeführt zu werden.
			b1.fork();
			b2.fork();
			b3.fork();
			
			

			// Join: Ergebnisse zusammenrechnen
			// Mit .join() liefert jeder Task seinen berechneten Wert
			// zurück. Die .join() Methode blockt entsprechend, bis
			// diesen Wert denn auch gibt.
			//
			// Vergleiche auch .get() welche bei Callables ein Future<T> liefert.
			
			int sum = b1.join() + b2.join() + b3.join();
			System.out.printf( //
			 "A collected sum=%d in %s%n", //
			 sum, //
			 Thread.currentThread().getName() //
			);
			
			System.out.println("TaskA compute beendet.");
			
			return sum;
		}
	}

	/**
	 * B: Eine ganz einfache Aufgabe, liefert id*10
	 * 
	 * Von RecursiveTask T können KEINE Objekte erzeugt werden; man kann diese
	 * Klasse allerdings ableiten. 
	 * 
	 * @author t2m
	 *
	 */
	static class TaskB extends RecursiveTask<Integer> {
		private final int id;

		/**
		 * Jeder Tasks hat eine id, damit wir diesen unterscheiden können.
		 * 
		 * @param id
		 */
		TaskB(int id) {
			this.id = id;
		}
		
	

		/**
		 * Jeder {@link RecursiveTask} fordert die compute() 
		 * Methode, damit in unserem Fall ein Task für den Typ Integer
		 * am Ende mit compute() auch einen Integer-Wert liefrt. 
		 */
		@Override
		protected Integer compute() {
			
			// Ausgabe, welcher B-Task hier gerade läuft.
			System.out.printf("Running B%d in %s%n",
			 id, Thread.currentThread().getName());
			return id * 10; // Berechnung multipliziert ID einfach mit 10.
		}
	}

	/**
	 * 
	 * @param args Keine Argumente
	 * @throws Exception
	 */
	public static void main(String[] args) throws Exception {
		
		// {#1} Erzeugt einen Work-Stealing-Pool mit (hier) 4 parallel arbeitenden Threads
		// {#2} Executors erzeugt stets Objekte einer Implementierungsklasse
	// {#3} Der Rückgabewert der Factory-Methoden ist aber ein Executor-Interface.
	// {#4} Wir müssen deshalb den Wert casten.
		//
		// WICHTIG: Dieser spezielle Pools ist in der Lage, verschachtelte
		// Tasks verwalten zu können.
		ForkJoinPool pool = (ForkJoinPool) Executors.newWorkStealingPool(4);

		// A wird in den Pool eingereicht
		// B1 bis B3 brauchen wir nicht übergeben, weil diese innerhalb von A
		// erzeugt werden. Der Task A ist also der Container für den Fork von 3
		// Sub-Tasks. 
		System.out.println("main: pool.submit(new TaskA() vorher");
		ForkJoinTask<Integer> taskA = pool.submit(new TaskA());
		System.out.println("main: pool.submit(new TaskA() nachher");

		// C: Holt das Gesamtergebnis ab
		// Einen Task "C" sparen wir uns weil dieser einfach der letzte
		// unserer main-Methode und damit dem Haupthread ist.
		System.out.printf("Final result in main: %d%n", taskA.get());
		
		// ! Man beachte, dass task.get() ein Future T liefert und damit zuerst einmal blockt is der Wert kommt. 

		// Abschließend unseren Executor für den Pool kontrolliert beenden.
		pool.shutdown();
	}
}


 

package com.stuelken.java.b4.parallel.c10.concurrentexecutor;

import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 
 * @param args
 */
public class ExecutorServiceDemo {

	public static void main(String[] args) {

		/*
		 * === Einfacher Executor
		 */

		Executor simpleExecutor = Executors.newSingleThreadExecutor();
		simpleExecutor.execute(() -> {
			System.out.println("Task mit einfachem Executor laeuft");
			try {
				Thread.sleep(2000);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		});

		((ExecutorService) simpleExecutor).isShutdown();

		/*
		 * === Vollwertiger ExecutorService
		 */

		ExecutorService poolExecutor = Executors.newFixedThreadPool(2);

		poolExecutor.submit(() -> {
			System.out.println("Task 1 via ExecutorService laeuft");
			try {
				Thread.sleep(1000);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		});

		poolExecutor.submit(() -> {
			System.out.println("Task 2 via ExecutorService laeuft");
			try {
				Thread.sleep(1000);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		});

		// Beispiel #3: Kontrolle via shutdown()
		System.out.println("main: sleep 10000!");
		try {
			Thread.sleep(10000);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		System.out.println("main: poolExecutor.shutdown!");
		poolExecutor.shutdown();

		System.out.println("main: Letzte Zeile. ");

		///////////////////////////////////
		// TODO: endet das Programm nicht?
		///////////////////////////////////
		if (false) {
			((ExecutorService) simpleExecutor).shutdown();
			System.out.println("main: Jetzt hat auch simple~ geendet. ");
		}

	}

}