Concurrency

Runnable t = () -> System.out.println("Hello");			// any returned value from a Runnable is ignored
new Thread(t).start();
System.out.println("World");

// the order of execution upon creation of a thread on line 1 depends upon the scheduler and can't be predicted beforehand
// can print "HelloWorld" or "WorldHello" because there are two threads - the one we created and one having main()

// start() creates a separate call stack for the thread execution
// calling start() implicitly calls run()
// Also if we use run() on line 2 above, it won't start a new thread and moreover it will just execute task in a non-concurrent synchronous way in main() thread only printing "HelloWorld"

// calling start() twice on same thread throws IllegalStateException, but we can call run() multiple times

new Thread(()->{ 
        for(int i = 0; i < 5; i++){
		  System.out.println(i);
        }
}).start();
new Thread(()->{
	System.out.println("A");
	System.out.println("B");
	System.out.println("C");
	System.out.println("D");
}).start();

// 3 threads - main() and 2 new threads we created
// output unkown until runtime, one possible outcome is:
/*
1
A
2
3
B
C
4
D
*/

// use t.join() to stop the current thread and wait for thread "t" to finish

Runnable task = () -> System.out.println("Hello");
Thread thread = new Thread(task);
thread.start();

thread.join();

System.out.println("World");

// output:
/* 
Hello
World
*/

// 1
class Foo implements Runnable{
    @Override
    public void run(){
        System.out.println("foo");
    }
}

// 2
class Bar extends Thread{
    @Override
    public void run(){
        System.out.println("bar");
    }
}

// in main()
Runnable task = new Foo();
Thread t1 = new Thread(task);
t1.start();

Thread t2 = new Bar();
t2.start();

var job = new Thread(() -> foobar());
job.setDaemon(true);		// create as daemon thread
job.start();				// we won't wait until foobar() finishes, we will exit once main() finishes

var job = new Thread(() -> taskMethod());

job.getState();			// prints thread state (RUNNABLE, WAITING, etc...)

Thread.sleep(1_000);	// sleeps for 1 sec, resumes thread execution upon timeout
// any other thread can interrupt a sleeping thread, in that case InterruptedException is thrown
// also, an already interrupted thread (isInterrupted() = true) can't sleep (throws InterruptedException)

var t2 = Thread.currentThread();	// get instance of current thread
t2.interrupt();			// used to throw InterruptedException in a sleeping thread

// calling interrupt() on a RUNNABLE thread (either self or from another thread) doesn't change its state (no halts or exceptions) but changes value on:
t2.isInterrupted();     // returns true

class Test extends Thread {
    @Override
    public void run() {
        System.out.println("A");
        var t = Thread.currentThread();
        System.out.println(t.isInterrupted());
        t.interrupt();		// self-interruption
        System.out.println("B");
        System.out.println(t.isInterrupted());
        System.out.println("C");
    }
}

public class Main {
    public static void main(String[] args) {
        var t = new Test();
        t.start();
        System.out.println("Z");
        System.out.println("Y");
        System.out.println("X");
    }
}

/*
Z
Y
X
A
false
B
true
C
*/

class Test extends Thread {
    @Override
    public void run() {
        System.out.println("A");

        var t = Thread.currentThread();
        try {
            Thread.sleep(5000);		// 5 sec
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        System.out.println("B");
        System.out.println("C");
    }
}

public class Main {
    public static void main(String[] args) {
        var t = new Test();
        t.start();
        t.interrupt();		// called immediately after thread start (< 5 sec)
        System.out.println("Z");
        System.out.println("Y");
        System.out.println("X");
    }
}

// InterruptedException is thrown

ExecutorService service = Executors.newSingleThreadExecutor();
try{
	service.execute(t1);	// t1, t2, and t3 are Runnable types (tasks)
	service.execute(t2);
	service.execute(t3);
}
finally{
	service.shutdown();
}

// this only creates two threads - one main() and one new thread. The intra-thread tasks are guaranteed to be executed sequentially in the order of submission and won't interfere with each other as was shown in the example at the beginning with Thread class.

try{
	service.execute(t1);
}
finally{
	service.shutdown();
}

service.awaitTermination(1, TimeUnit.MINUTES);
if(service.isTerminated()){
	System.out.println("Done!");
}
else{
	System.out.println("Not completed yet!");
}

Future<?> result = service.submit(t1);		// Future ref's formal type is whatever type t1 returns

// instance methods of Future<>
result.isDone()						// returns true if completed, exception or cancelled 
result.isCancelled()				// returns true if cancelled
result.cancel()						// attempts to cancel and returns true if successfully cancelled
result.get()						// get result or wait endlessly
result.get(10, TimeUnit.SECONDS)	// get result or wait for specified time, throw TimeoutException upon timeout

Future<Integer> result = service.submit(() -> 30 + 11);
System.out.println(result.get()); 	// 41

// todo the above with Runnable and execute() will be too complex as we will need a common variable and polling with sleep or an interrupt. 

ScheduledExecutorService service = Executors.newSingleThreadScheduledExecutor();

Runnable task1 = () -> System.out.println("Hello Zoo");
Callable<String> task2 = () -> "Monkey";
ScheduledFuture<?> r1 = service.schedule(task1, 10, TimeUnit.SECONDS);
ScheduledFuture<?> r2 = service.schedule(task2, 8, TimeUnit.MINUTES);

// The first task is scheduled 10 seconds in the future, whereas the second task is scheduled 8 minutes in the future
// if service is shutdown before they actually get to execute, then they will be discarded

service.scheduleAtFixedRate(task, 5, 1, TimeUnit.MINUTES);	
// submits task after initial delay of 5 mins, and keeps submitting after 1 min intervals regardless of previous task finished
// be careful as the tasks can pile up if period is less than execution time of the task

service.scheduleWithFixedDelay(task1, 0, 2, TimeUnit.MINUTES);
// the 2 min delay is after the finish of the previous task unlike above method

ExecutorService 
newSingleThreadExecutor()

ScheduledExecutorService 
newSingleThreadScheduledExecutor()

ExecutorService 
newCachedThreadPool()
/* Creates thread pool that creates new threads as 
needed but reuses previously constructed threads 
when they are available. */

ExecutorService 
newFixedThreadPool(int)
/* Creates thread pool that reuses fixed number of 
threads operating off shared unbounded queue. */

ScheduledExecutorService 
newScheduledThreadPool(int)
/* Creates thread pool that can schedule commands 
to run after given delay or execute periodically. */

int x = 0;
public void incrementX(){
	Syste.out.println(++x);
}

ExecutorService service = Executors.newFixedThreadPool(20);
for(int i = 0; i < 10; i++){
	service.submit(() -> incrementX());
}

// can print in any order; even duplicates, because 10 threads will access x and overwrites will happen

volatile int x = 0;

AtomicInteger x = new AtomicInteger(0);
x.incrementAndGet();	// equivalent to ++x

// other methods
x.set(9);	// assignment
x.getAndIncrement();	// x++

// on an Object - block
synchronized(this) {  }

// on method
synchronized void foobar() {  }

-----------------------------------

// in static method
synchronized(Foobar.class) {  }		 // since we can't use "this" object here; so we use class object

Test t = new Test();
synchronized(t) {  }                 // create and use any random instance

// on static method
static synchronized void foobar() {  }

Lock lock = new ReentrantLock();
try{
	lock.lock();		// waits/blocks indefinitely if other thread has lock
	// code
}
finally{
	lock.unlock();
}

// above code in synchronized block
Object obj = new Object();
synchronized(obj){
	// code
}

// constructor of ReentrantLock
Lock lock = new ReentrantLock(true);		// fairness property

// true - grant lock in order it was requested
// false - default no-arg call

// attempting to unlock a lock which was never acquired leads to
IllegalMonitorStateException

// code demo
public static void printHello(Lock lock) {
    try {
 	  lock.lock();
 	  System.out.println("Hello");
    } 
    finally {
 	  lock.unlock();
    } 
}

public static void main(String[] args) {
	Lock lock = new ReentrantLock();
	new Thread(() -> printHello(lock)).start();		// sending lock to be acquired by a thread
	if(lock.tryLock()) {							// tryLock() attempting to lock to Main thread
 		try {
 			System.out.println("Lock obtained, entering protected code");
 		} 
 		finally {
 			lock.unlock();
 		}
	} else {
 		System.out.println("Unable to acquire lock, doing something else");
	}
}

// a counting semaphore
Semaphore smp = new Semaphore(3);       
// permits 3 threads to acquire it simultaneously; others get blocked in a queue

public static void printHello(Semaphore smp) {
    try {
        smp.acquire();
        // code
    } 
    finally {
        smp.release();
    } 
}

var c1 = new CyclicBarrier(5);

// in code executed by multiple threads
c1.await()

// keeps waiting until 5 threads are in waiting status, then resumes on all

// throws
InterruptedException
BrokenBarrierException

// if we have 15 threads and barrier size is 5 then -
// await 5 threads and break barrier
// await next 5 and break barrier
// await next 5 and break barrier
// so its called three times, and everytime threads resume in set of 5

// in main method
CountDownLatch countDownLatch = new CountDownLatch(5);      // create latch

foobar(1, countDownLatch);

// call foobar() and wait for all tasks to complete
countDownLatch.await();


// task method
void foobar(int x, CountDownLatch countDownLatch){
    // do some processing and after that dec counter
    countDownLatch.countDown();
}

// non-concurrent collection
Map<String, String> mp = new HashMap<>();

// use a concurrent collection - multiple threads at a time can modify
Map<String, String> cmp = new ConcurrentHashMap<>();

// or make an existing one synchronized - single thread at a time can modify
Collections.synchronizedMap(mp);

// or make an existing one immutable - none can modify
Collections.unmodifiableMap(mp);

// first thread enters this; after acquiring lock for obj1
synchronized (obj1){
    synchronized (obj2){  }
}

// second thread enters this; after acquiring lock for obj2
synchronized (obj2){
    synchronized (obj1){  }
}

// both keep waiting for each other to release locks on thier respective nested sync block's obj
// so that they can acquire lock and enter nested sync block but it never happens - circular waiting

// making a Stream parallel
collection.stream().parallel();
collection.parallelStream();

// checking if a stream is parallel
stream.isParallel();

stream.unordered();

List<StockInfo> getStockInfo(Stream<String> companies) {
    return companies.parallel()
        .map(this::getStockInfo)        // slow operation; calls method on every element
        .collect(Collectors.toList());
}

// this hangs (blocks) every other method in execution that is using parallel streams if enough threads aren't available in the pool
// not only "companies" stream, but any parallel stream part of the code flow!

Runnable task = () -> System.out.println("Hello");

// using Thread class
Thread virtualThread = Thread.startVirtualThread(task);

// another way
Thread thread = Thread.ofVirtual().unstarted(task);
thread.start();


// using concurreny API
ExecutorService service = Executors.newVirtualThreadPerTaskExecutor();
// creates new virtual thread for each submitted task


thread.isVirtual();      // tells if the thread is virtual

// perform a computation async - takes in a Supplier
CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(() -> "Hello");
// get value from future instance; it is a blocking call as it waits for return value
completableFuture.get();

// explicitly complete the future and get value (can be used in error handling scenarios)
completableFuture.complete("Done!");


// to process the result of a computation - feed it to a function or consumer (chaining)
// note that these are non-blocking just like supplyAsync() and do not block the calling thread

// feed result of async computation into a Function
CompletableFuture<String> future = completableFuture.thenApply(s -> s + " World");

// feed to a Consumer and it returns a Void placeholder type 
CompletableFuture<Void> future = completableFuture.thenAccept(s -> System.out.println("Computation returned: " + s));

// nothing to feed, nothing to get back; print a line in the console on future.get() call 
CompletableFuture<Void> future = completableFuture.thenRun(() -> System.out.println("Computation finished."));


// provide a callback to run upon completion using whenComplete method (non-blocking)
completableFuture.whenComplete((result, exception) -> {
    if (exception == null) {
        // the computation completed normally
        System.out.println("Result: " + result);
    } else {
        // the computation completed with an exception
        System.err.println("Exception: " + exception.getMessage());
    }
});


// avoid doing a blocking operation like get() where you don't want to wait. Use chaining and callbacks to process async computation's result.

// blocking operation
CompletableFuture<Void> combinedFuture  = CompletableFuture.allOf(cf1, cf2, cf3);

ExecutorService exec = Executors.newFixedThreadPool(20);
CompletableFuture<String> comFut = CompletableFuture.thenSupplyAsync(task, exec);

CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
    // System.out.println(Thread.currentThread().isDaemon());
    Thread.sleep(2000);
    System.out.println("A");
    Thread.sleep(2000);
    System.out.println("B");
    Thread.sleep(2000);
    System.out.println("C");
});

System.err.println("Exiting...");

// Output: "Exiting..." and then process finishes with exit code 0!

// Reason: the async thread is from ForkJoinPool and its a daemon thread. If main thread exits first, JVM forces daemon threads to terminate too.

CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
        // System.out.println(Thread.currentThread().isDaemon());
    Thread.sleep(2000);
    System.out.println("A");
    Thread.sleep(2000);
    System.out.println("B");
    Thread.sleep(2000);
    System.out.println("C");
});

System.err.println("Exiting...");
// Thread.sleep(10000);
future.join();

/* Output: 
Exiting...
A
B
C
*/