Последовательность «фаз» синхронизирована с семафорами, которые не работают должным образом

Question

Я изучаю потоки и синхронизацию и пытаюсь настроить систему «фазы», ​​в которой первая фаза генерирует сообщение (для удобства int), а затем передает его фазе 2, которая изменяет это (умножает это на 2), и затем передает это к последней фазе, которая далее изменяет это и печатает это к консоли. Проблема в том, что третий этап никогда не запускается, даже если он получает сообщения.

В моем примере я установил массив семафоров, называемых «ресурсами», в которых семафор A (индекс 0) имеет 4 разрешения, семафор B - 3, а C - 2. Все установлены как справедливые.

Я пытался настроить Семафоры на справедливость, но это не решило мою проблему. Я также пытался изменить время сна, но мне не повезло.

class Fase1 extends Thread{
    private int i = 0;
    private Semaphore[] resources;
    private Fase2 recipient;

    public Fase1(Semaphore[] res, Fase2 fase2){
        recipient=fase2;
        resources=res;
    }

    @Override
    public void run(){
        try{
            while(true){
                resources[0].acquire(2);
                resources[1].acquire(2);
                recipient.receiveMessage(i);
                i++;
                sleep(200);
                resources[1].release(2);
                resources[0].release(2);
            }
        } catch (InterruptedException e){
        }
    }
}

class Fase2 extends Thread{
    private Semaphore[] resources;
    private Fase3 recipient;
    private boolean receivedMessage = false;
    private int message = 0;

    public Fase2(Semaphore[] res, Fase3 fase3){
        recipient=fase3;
        resources=res;
    }

    @Override
    public void run(){
        try{
            while(true){
                if(receivedMessage){
                    resources[0].acquire(2);
                    resources[1].acquire(2);
                    resources[2].acquire(2);
                    recipient.receiveMessage(message*2);
                    receivedMessage = false;
                    sleep(200);
                    resources[2].release(2);
                    resources[1].release(2);
                    resources[0].release(2);
                }
            }
        } catch (InterruptedException e){
        }
    }
    public void receiveMessage(int msg){
        message = msg;
        receivedMessage = true;
    }
}

class Fase3 extends Thread{
    private Semaphore[] resources;
    private boolean receivedMessage = false;
    private int message = 0;

    public Fase3(Semaphore[] res){
        resources=res;
    }

    @Override
    public void run(){
        try{
            while(true){
                if(receivedMessage){
                    resources[1].acquire(2);
                    resources[2].acquire(2);
                    System.out.println(message+1);
                    receivedMessage = false;
                    sleep(200);
                    resources[2].release(2);
                    resources[1].release(2);
                }
            }
        } catch (InterruptedException e){
        }
    }
    public void receiveMessage(int msg){
        message = msg;
        receivedMessage = true;
    }
}

Я заметил, что разрешения каким-то образом облажались, это похоже на то, что некоторые потоки не выпускают их должным образом, хотя мне это кажется правильным.

Answer 1

В вашем дизайне есть фундаментальный недостаток: вы синхронизируете доступ к ресурсу message, но не к флагу receivedMessage.Когда вы устанавливаете флаг в true в потоке # 2, JVM не обязана распространять эту запись в поток # 3, поскольку этот поток не выполняет синхронизацию, пока он не окажется внутри блока if, что вполне может никогда не произойти.То же самое касается связи между потоками № 1 и № 2.

Операции acquire() и release() действуют как точки синхронизации и делают записи видимыми в потоках, поэтому вам необходимопозвоните им, прежде чем проверять флаги.Например, в Fase3:

    @Override
    public void run(){
        try{
            while(true){
                resources[1].acquire(2); // All writes by thread #2 are now visible
                if(receivedMessage){
                    resources[2].acquire(2);
                    System.out.println(message+1);
                    sleep(200);
                    receivedMessage = false;
                    resources[2].release(2);
                }
                resources[1].release(2);
            }
        } catch (InterruptedException e){
        }
    }

Существует еще одно решение, заключающееся в создании receivedMessage flags volatile, но правильнее использовать механизм единой блокировки.

В качестве дополнительного примечания лучше использовать Runnable с вместо расширения Thread.