package com.stuelken.java.b4.parallel.c40.synchronizer;

import java.util.concurrent.CountDownLatch;

/**
 * @author t2m
 */
public class CountDownLatchExample {
	
	public static void main(String[] args) throws InterruptedException {
		new CountDownLatchExample().doCountDownLatch();
	}
	
	/**
	 * Use Case: Warten auf mehrere Vorbereitungen (z. B. Game-Assets,
	 * Spieler-Eingaben)
	 * @throws InterruptedException 
	 */
	public void doCountDownLatch() throws InterruptedException {

		CountDownLatch latch = new CountDownLatch(3);

		Runnable worker = () -> {
			System.out.println("Worker "+ Thread.currentThread().getName()+" gestartet");
			latch.countDown();
		};

		for (int i = 0; i < 10; i++) {
			new Thread(worker, "T "+i).start();
		}

		latch.await(); // blockiert bis alle Threads durch sind
		System.out.println("Alle Worker fertig");

	}
}
 

package com.stuelken.java.b4.parallel.c40.synchronizer;

import java.util.Random;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;

/**
 * @author t2m
 */
public class CyclingBarrierExample {

	public static void main(String[] args) {
		doCyclingBarrier();
	}

	/**
	 * Use Case: Alle Spieler müssen erst fertig sein, bevor die Runde fortschreitet
	 */
	public static void doCyclingBarrier() {

		int parties = 3;

		// Die Aktion, welche die CyclingBarrier ausführt.
		Runnable barrierAction_A = () -> {
			System.out.println("Auf die Plaetze!");
		};
		Runnable barrierAction_B = () -> {
			System.out.println("Fertig ... ");
		};
		Runnable barrierAction_C = () -> {
			System.out.println("Los!!!");
		};

		// Instanz erzeugten mit Anzahl der Akteure und der Aktion.
		CyclicBarrier barrierA = new CyclicBarrier(parties, barrierAction_A);
		CyclicBarrier barrierB = new CyclicBarrier(parties, barrierAction_B);
		CyclicBarrier barrierC = new CyclicBarrier(parties, barrierAction_C);

		// Jeder Spieler wird statt als Klasse kurzerhand als
		// Runnable einzig und allein mit einer Funktion
		// gestartet.

		Runnable player = () -> {
			String i = Thread.currentThread().getName();
			try {
				System.out.println("Spieler " + i + " wartet auf A");
				barrierA.await();
				System.out.println("Spieler " + i + " wartet auf B");
				barrierB.await();
				System.out.println("Spieler " + i + " wartet auf C");
				barrierC.await();
				Random r = new Random();
				for (int j = 0; j < 100; j = j + 10) {
					System.out.println("Spieler " + i + " rennt und erreicht " + j + "m");
					Thread.sleep(r.nextInt(1000) + 2000);
				}
				System.out.println("Spieler " + i + " kommt ins Ziel!");
			} catch (InterruptedException | BrokenBarrierException e) {
				e.printStackTrace();
			} // blockiert
			System.out.println("Spieler in Thread " + i + " jubelt!");
		};

		/*
		 * 3 Threads für je einen Player-Runnable.
		 */
		for (int i = 0; i < 3; i++) {
			Thread t = new Thread(player);
			t.setName("T" + i);
			t.start();
		}

		/*
		 * @todo: Berechne als Ergebnis des Programms, wer das Rennen am Ende gewonnen
		 * hat, indem nur Klassen/Interfaces aus java.concurrent statt Threads genutzt
		 * werden.
		 */

		/*
		 * @todo: Gebe aus, wer gerade im Verlauf des Rennens die Führung hat.
		 */

	}

} 

package com.stuelken.java.b4.parallel.c40.synchronizer;

import java.util.concurrent.*;

public class PhaserHelper {

 public void doPhaser() throws InterruptedException, ExecutionException {
 	
 	int parties = 1;
 Phaser phaser = new Phaser(parties); // Hauptthread registriert

 Callable<String> phaseTask = () -> {
 phaser.register();
 for (int i = 1; i <= 4; i++) {
 System.out.println(Thread.currentThread().getName() + " Phase " + i + " gestartet");
 phaser.arriveAndAwaitAdvance(); // auf nächste Phase warten
 }
 return Thread.currentThread().getName() + " ist fertig.";
 };

 ExecutorService pool = Executors.newFixedThreadPool(3);
 Future<String> f1 = pool.submit(phaseTask);
 Future<String> f2 = pool.submit(phaseTask);
 Future<String> f3 = pool.submit(phaseTask);

 // Hauptthread nimmt ebenfalls an 4 Phasen teil:
 for (int i = 1; i <= 4; i++) {
 System.out.println("Main thread Phase " + i + " beendet");
 phaser.arriveAndAwaitAdvance();
 }

 // Ausgabe von Ergebnissen (damit main bis zum Ende wartet):
 System.out.println(f1.get());
 System.out.println(f2.get());
 System.out.println(f3.get());

 pool.shutdown();
 }

 public static void main(String[] args) throws InterruptedException, ExecutionException {
 new PhaserHelper().doPhaser();
 }
}


 

package com.stuelken.java.b4.parallel.c40.synchronizer;

import java.util.concurrent.Exchanger;

/**
 * @author t2m
 */
public class ExchangerExample {
	
	public static void main(String[] args) {
		new ExchangerExample().doExchanger();
	}

	/**
	 * 
	 */
	public void doExchanger() {
		Exchanger<String> exchanger = new Exchanger<>();

		new Thread(() -> {
			try {
				// Wartet auf einen anderen Thread.
				String response = exchanger.exchange("Ping from A");
				System.out.println(Thread.currentThread().getName()+" Erhielt: " + response);
			} catch (InterruptedException e) {
			}
		},"A").start();

		new Thread(() -> {
			try {
				// Wartet auf einen anderen Thread.
				String response = exchanger.exchange("Pong from B");
				System.out.println(Thread.currentThread().getName()+" Erhielt: " + response);
			} catch (InterruptedException e) {
			}
		},"B").start();

	}
}
 

package com.stuelken.java.b4.parallel.c40.synchronizer;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;

/**
 * Semaphore dienen als Synchronizer für Rate Limiting, begrenzter
 * Ressourcenzugriff.
 * 
 * 
 * @author t2m
 *
 */
public class SemaphoreHelper {

	public static class MySemaphore {
	}

	public static void main(String[] args) {
		runExample();
	}

	/**
	 * 
	 */
	public static void runExample() {

		ExecutorService executor = Executors.newFixedThreadPool(10);
		
		// FALSCH!!!!
		// CountDownLatch latch = new CountDownLatch(10);
		// Jeder Latch wird nur 1x ausgeführt! Wir wollen
		// aber zwei Gruppen. 
		
		CountDownLatch latchUnfair = new CountDownLatch(10);
		CountDownLatch latchFair = new CountDownLatch(10);		

		/**
		 * Creates a Semaphore with the given number of permits and nonfair fairness
		 * setting. Parameters:permits the initial number of permits available.This
		 * value may be negative, in which case releasesmust occur before any acquires
		 * will be granted.
		 */
		Semaphore semaphoreNotFair = new Semaphore(2, false);
		Semaphore semaphoreFair = new Semaphore(2, true);

		/**
		 * 
		 */
		Runnable accessUnfair = () -> {

			try {
				/*
				 * {#1} Statt eines Monitor-Objekts wird ein Semaphore-Synchronizer verwendet.
				 * 
				 * Acquires a permit from this semaphore, blocking until one is available.
				 *
				 * <p>Acquires a permit, if one is available and returns immediately, reducing
				 * the number of available permits by one.
				 *
				 * <p>If no permit is available then the current thread becomes disabled for
				 * thread scheduling purposes and lies dormant until some other thread invokes
				 * the {@link #release} method for this semaphore and the current thread is next
				 * to be assigned a permit.
				 *
				 * <p>If the current thread is {@linkplain Thread#interrupt interrupted} while
				 * waiting for a permit then it will continue to wait, but the time at which the
				 * thread is assigned a permit may change compared to the time it would have
				 * received the permit had no interruption occurred. When the thread does return
				 * from this method its interrupt status will be set.
				 */
				System.out.println(Thread.currentThread().getName() + " wartet...");

				semaphoreNotFair.acquire(); // WICHTIG!

				System.out.println(Thread.currentThread().getName() + " darf rein");

				Thread.sleep(1000);

			} catch (InterruptedException e) {
			} finally {
				/*
				 * Releases a permit, returning it to the semaphore.
				 *
				 * <p>Releases a permit, increasing the number of available permits by one. If
				 * any threads are trying to acquire a permit, then one is selected and given
				 * the permit that was just released. That thread is (re)enabled for thread
				 * scheduling purposes.
				 *
				 * <p>There is no requirement that a thread that releases a permit must have
				 * acquired that permit by calling {@link #acquire}. Correct usage of a
				 * semaphore is established by programming convention in the application.
				 */
				System.out.println(Thread.currentThread().getName() + " released die Semaphore.");

				semaphoreNotFair.release(); // WICHTIG!
				
				// Wenn wir einen Latch verwenden, müssen wir dafür sorgen, dass
				// der Latch auch etwas zu zählen hat. Anderenfalls weiß
				// Latch nicht, wann alle Tasks abgearbeitet sind, und latch.await()
				// endet nicht. 
				latchUnfair.countDown(); // WICHTIG!				
				
			}
		};

		/**
		 * === Definition eines Runnables für "fair".
		 */
		Runnable accessFair = () -> {
			

			
			try {
				System.out.println(Thread.currentThread().getName() + " wartet...");
				semaphoreFair.acquire(); // WICHTIG!
				System.out.println(Thread.currentThread().getName() + " darf rein");
				Thread.sleep(1000);
			} catch (InterruptedException e) {
			} finally {
				System.out.println(Thread.currentThread().getName() + " released die Semaphore.");
				semaphoreFair.release(); // WICHTIG! Nicht Fair/Unfair verwechseln!
				
				// Wenn wir einen Latch verwenden, müssen wir dafür sorgen, dass
				// der Latch auch etwas zu zählen hat. Anderenfalls weiß
				// Latch nicht, wann alle Tasks abgearbeitet sind, und latch.await()
				// endet nicht. 
				latchFair.countDown(); // WICHTIG!

			}
		};

		System.out.println("Unfair:");

		for (int i = 0; i < 10; i++) {
			new Thread(accessUnfair).start();
		}

		try {
			latchUnfair.await();

			System.out.println("\nUNFAIR-Fertig, jetzt FAIR...");
		} catch (InterruptedException e) {
			Thread.currentThread().interrupt();
		}

		System.out.println();
		System.out.println("Fair:");
		for (int i = 0; i < 10; i++) {
			new Thread(accessFair).start();
		}

		try {
			latchFair.await();
			System.out.println("\nFAIR-Fertig.");
		} catch (InterruptedException e) {
			Thread.currentThread().interrupt();
		}

		// Beenden.
		executor.shutdown();

	}

}
 

package com.stuelken.java.b4.parallel.c40.synchronizer;

import java.util.concurrent.*;
import java.util.*;
import java.util.stream.*;

/**
 * 
 * @author t2m
 */
public class SemaphoreCompletableFutureDemo {

	public static void main(String[] args) {

		Semaphore semaphoreUnfair = new Semaphore(2, false);
		Semaphore semaphoreFair = new Semaphore(2, true);
		ExecutorService executor = Executors.newFixedThreadPool(10);

		// Erst unfair laufen lassen
		List<CompletableFuture<Void>> unfairTasks = IntStream.range(0, 10)
		 .mapToObj(
		 i -> CompletableFuture.runAsync(() -> task("Unfair-" + i, semaphoreUnfair), executor))
		 .collect(Collectors.toList());

		// Wenn alle unfair fertig → dann fair starten
		CompletableFuture
		 .allOf(unfairTasks.toArray(new CompletableFuture[0]))
		 .thenRun(() -> {
			 System.out.println("\nAlle Unfair-Tasks fertig. Starte Fair-Phase...\n");
			 List<CompletableFuture<Void>> fairTasks = IntStream.range(0, 10)
			 .mapToObj(
			 i -> CompletableFuture.runAsync(() -> task("Fair-" + i, semaphoreFair), executor))
			 .collect(Collectors.toList());

			 CompletableFuture
			 .allOf(fairTasks.toArray(new CompletableFuture[0]))
			 .thenRun(() -> {
				 System.out.println("\nAlle Fair-Tasks abgeschlossen.");
				 executor.shutdown();
			 });
		 });
	}

	static void task(String name, Semaphore semaphore) {
		try {
			System.out.println(name + " wartet...");
			semaphore.acquire();
			System.out.println(name + " hat Zugriff");
			Thread.sleep(1000);
		} catch (InterruptedException e) {
			Thread.currentThread().interrupt();
		} finally {
			semaphore.release();
			System.out.println(name + " gibt Zugriff frei");
		}
	}
}