Synchronizacją wątków za pomocą bloków synchronized

Cześć, zagłębiam się w Jave i wątki. Cel jest taki, aby 1 wątek inkrementował zmienną a drugi dekrementował zmienną. Problem tutaj jest z synchronizacją. Generalnie wynik powinien być mniej więcej:
Wątek 1: 1
Wątek 2: 0
Wątek 1: 1
Wątek 2: 0
Wątek 1: 1
...
...
Próbuję to zadanie wykonać za pomocą wait() i notify(), niestety wynik jest błędny (albo zapada blokada w 2 wątkach albo wyświetla się błędna kolejność).

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

public class MainTest
{
	public static volatile int a = 0;
	public static void main(String[] args)
	{
		Runnable r1 = () -> 
		{
			Thread.currentThread().setName("Watek 1");
			for(int i = 0; i < 5; i++)
			{
				a++;
				System.out.println("A1x = " + a );
				synchronized(MainTest.class)
				{
					if(a > 0)
					{
						try 
						{
							System.out.println(Thread.currentThread().getName() + " is waiting...");
							MainTest.class.wait();
						} 
						catch (InterruptedException e) 
						{
							e.printStackTrace();
						}
					}
				}
			}
		};
		Runnable r2 = () ->
		{
			try {
				Thread.sleep(100);
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
			Thread.currentThread().setName("Watek 2");
			for(int i = 0; i < 5; i++)
			{	
				synchronized (MainTest.class) 
				{
					if(a > 0)
					{
						System.out.println("WAKE UP");
						a--;
						System.out.println("A2x = " + a );
						MainTest.class.notify();
					}
				}
			}
		};
		
		ExecutorService exec = Executors.newCachedThreadPool();
		exec.submit(r1);
		exec.submit(r2);
		
		exec.shutdown();
		try 
		{
			exec.awaitTermination(1, TimeUnit.DAYS);
		} catch (InterruptedException e) 
		{
			///
		}
	
	
	}
}

Jak sobie wyobrażam, że to powinno działać:

: A
W r1 następuje a++ [a = 1], wchodzi w blok synchronized i w ifa wywołuje wait().
w r2 wchodzi w blok synchronized i ifa, następuje a-- [a = 0] następnie wywoływane jest notify, budzi się r1.
goto A

Troszkę tam trzeba logikę poprawić. Natomiast zanim do tego przejdę, krótkie wyjaśnienie problemu na przykładzie:

Przykładowy output:

gynvael:haven-windows> java MainTest
A1x = 1
Watek 1 is waiting...
WAKE UP
A2x = 0
A1x = 1
Watek 1 is waiting...

Na samym końcu program "się zawiesił", a konkretniej stanął na .wait() r1.

Wyjaśnienie:

1. Oba wątki są odpalane. Załóżmy, że nawet w tej samej kolejności, choć jest to nieistotne, ponieważ r2 od razu uśnie na Thread.sleep(100), więc wykona się PRAWDOPODOBNIE (o ile system nie jest zbyt obciążony i scheduler tak zadecyduje) wątek r1 najpierw.

2. Wątek r1 faktycznie zaczyna się wykonywać i wypisuje
   A1x = 1
   Po czym łapie mutex (synchronized), wypisuje:
   Watek 1 is waiting...
   I idzie spać:
   MainTest.class.wait(); (to zwalnia mutex btw)

3. Chwilę później scheduler budzi r2, i ten wchodzi do pętli, sprawdza warunek a > 0 (jak wiadomo ten warunek jest w tym momencie spełniony), po czym wypisuje:
   WAKE UP
   Zmniejsza a, i wypisuje:
   A2x = 0
   Po czym robi MainTest.class.notify() i zwalnia mutex.

Iii... w tym momencie pewnie byś się spodziewał, że wątek r1 się obudzi i... Ale to tak nie działa ;)
Wątek r1 jest obecnie w stanie uśpionym i zostanie faktycznie obudzony dopiero gdy scheduler uzna to za stosowne - załóż, że to jest w pełni niedeterministyczne.

Natomiast wątek r2 jest jak bardziej obudzony, więc wykonuje się dalej, tj. robi kolejny obrót tej pętli:

  for(int i = 0; i < 5; i++)
            {    
                synchronized (MainTest.class) 
                {
                    if(a > 0)
                    {
                       // Ten warunek już nie jest spełniony, bo r1 śpi, a a==0.
                    }
                }
            }

Ta pętla wykona się oczywiście błyskawicznie, więc r2 zakończy działanie i wyjdzie.

4. Chwilę później scheduler w końcu obudzi r1 - czy to z losowego powodu (to się zdarza w przypadku zmiennych warunkowych), czy ponieważ notify zostało wywołane.
   Zacznie on więc wykonywać pętle dalej, tj. zrobi a++ i wypisze:
   A1x = 1
   Po czym zajmie mutex, wypisze:
   Watek 1 is waiting...
   I zaśnie wiecznym snem w wait(), ponieważ wątek r2 już wyszedł i nie będzie kto miał obudzić wątku r1.
   Przy czym, może nastąpić spątaniczne wybudzenie - w takim wypadku wątek zrobi kolejną iteracje pętli i znowu zaśnie, zwiększając a do kolejnych wartości (2, 3, 4, 5). Aż w końcu wyjdzie.

I tak to wygląda.

W związku z tym musisz poprawić logikę w taki sposób, by:

1. Wątek r2 powiadamiał wątek r1 kiedy ten może się wykonać (to już masz).
2. Wątek r1 powiadamiał wątek r2 kiedy ten może się w końcu wykonać (tego brakuje; wątek r2 nigdy nie robi żadnego wait; tym samym będziesz mógł się pozbyć tego sleep(100), które i tak nie zawsze działa tak jak byś chciał - scheduler może dłużej uruchamiać wątek r1 niż 100ms przy obciążonym systemie).
3. Idealnie, warto żebyś dodał jakieś warunki testujące, czy wait() nie wybudził się spontanicznie. Zaleceniem w Java (jak i w innych językach dysponujących zmiennymi warunkowymi) jest korzystanie z wait() w pętli, która sprawdza jakiś tam warunek (u Ciebie tym warunkiem dla r1 mogło by być while(a!=0) wait(), natomiast dla r2 while(a==0) wait()).

Co więcej, blok synchronized MUSI obejmować inkrementacje a++ - na tym polega synchronizowanie dostępu do zmiennej dzielonej, że dostęp jest faktycznie synchronizowany ;) (obecnie w r1 masz dostęp do zmiennej "a" poza blokiem synchronized).

OK, to tak na szybko. Jesteś blisko rozwiązania ;)

Dziękuję za wytłumaczenie, teraz mój kod wygląda tak i działa tak jak chciałem :) :

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

class ThreadIncDec 
{
	private volatile int a = 0;
    public synchronized void inc() 
    {
        while(a != 0) 
        {
            try 
            {
                wait();
            }
            catch (InterruptedException e) {}
        }
        a++;
        System.out.println("R1 = " + a);
        notify();
    }

    public synchronized void dec() 
    {
        while(a == 0) 
        {
            try 
            { 
                wait();
            } 
            catch (InterruptedException e) {}
        }
        a--;
        System.out.println("R2 = " + a);
        notify();
    }
}

public class MainTest
{
	public static ThreadIncDec tObj = new ThreadIncDec();
	public static void main(String[] args) 
	{
		Runnable r1 = () ->
		{
			for(int i = 0; i < 5; i++)
				tObj.inc();	
		};
		
		Runnable r2 = () ->
		{
			for(int i = 0; i < 5; i++)
				tObj.dec();	
		};
		
		ExecutorService exec = Executors.newCachedThreadPool();
        exec.submit(r1);
        exec.submit(r2);
 
        exec.shutdown();
        try 
        {
            exec.awaitTermination(1, TimeUnit.DAYS);
        } 
        catch (InterruptedException e) { }
		
	}
}

Będą problemy przy przedwczesnym przerwaniu, powinno być tak:

        try 
        {
            while(a!=0) wait();
            System.out.println("R1 = "+(a++));
            notify();
        }
        catch (InterruptedException e) {}

dec - podobnie