java中的简单屏障同步

Question

我试图理解障碍问题。我对编程仍然很陌生，但在课堂上遇到了这个问题要解决。

“我必须使用计数信号量来解决屏障问题。你可以假设有一个共享变量 N 表示系统中的并发线程数。当前 N -1 个线程到达屏障时，它们应该阻塞直到第 N 个线程到达，此时所有线程都可能继续。

共享计数器变量可用于跟踪已到达的线程数，信号量互斥锁和屏障可用于解决同步问题。”

import java.util.concurrent.Semaphore;
public class BarrierSynchronization extends Thread {

int N;
int count;
Semaphore mutex;
Semaphore barrier;

public BarrierSynchronization ()
{
    this.N = 5;
    this.count = 0;
    this.mutex = new Semaphore(1);
    this.barrier = new Semaphore(0);
}

public void run()
{
    try {
        mutex.acquire();  
        count = count + 1;
        System.out.println(Thread.currentThread().getName() + ": " + count);
        mutex.release();

        if (count == N)
        {
            barrier.release();
            System.out.println("All " + count + " threads have reached the barrier. The barrier is now open" );
        } // unblock one thread

            barrier.acquire();
            barrier.release();  
            System.out.println(Thread.currentThread().getName() + " has passed the barrier");


    } catch (InterruptedException e) {
        e.printStackTrace();
    }
}   

}

我尝试实现信号量小书中的伪代码。我在主类中调用了这个线程并运行了它，但是由于某种原因它给出了关于 mutex.wait() 的错误。当我删除它运行但什么也不显示的代码时。我究竟应该为这个问题做些什么？

public class Main {

public static void main(String[] args) throws InterruptedException 
{           
    BarrierSynchronization barrier = new BarrierSynchronization();

    Thread bs1 = new Thread(barrier);
    Thread bs2 = new Thread(barrier);
    Thread bs3 = new Thread(barrier);
    Thread bs4 = new Thread(barrier);
    Thread bs5 = new Thread(barrier);

    bs1.start();
    bs2.start();        
    bs3.start();        
    bs4.start();
    bs5.start();
}

why does it output the rare one before the barrier is unlocked for all the threads? i think im close enough to solving this problem. is a race condition or something? CLICK TO SEE IMAGE

Answer 1

可能有点晚了，但这里有一个驱动程序代码有效的实现。您必须保证互斥并跟踪到达屏障的线程数。

public class Barrier {
    private int capacity;
    private Semaphore s, exclusao, counter;

    public Barrier(int capacity) {
        this.capacity = capacity;
        counter = new Semaphore(0);
        s = new Semaphore(0);
        exclusao = new Semaphore(1);
    }

    public void espera() throws InterruptedException {
        exclusao.acquire();
        if (counter.availablePermits() < capacity - 1) {
            counter.release();
            exclusao.release();

            s.acquire();
        } else {            
            exclusao.release();

            System.out.println("RELEASE ALL");
            for (int i = 0; i < capacity; i++) {
                s.release();
            }
        }
    }

}

class TesteThread extends Thread {
    private Barrier b;
    private long waitPeriod;

    public TesteThread(long wait, Barrier b) {
        this.b = b;
        this.waitPeriod = wait;
        System.out.println("Thread started" + this.getName());

    }

    public void espera() throws InterruptedException {
        b.espera();
    }

    @Override
    public void run() {
        try {
            System.out.println("Thread a dormir " + this.getName());
            sleep(waitPeriod);
            System.out.println("Thread a esperar " + this.getName());
            espera();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

class BarrierExample {

    public static void main(String[] args) throws InterruptedException {
        Barrier BR = new Barrier(5);

        TesteThread[] teste = new TesteThread[5];
        for (int i = 0; i < teste.length; i++) {
            teste[i] = new TesteThread((long) (Math.random() * 1000), BR);
            teste[i].start();
        }
        for (int i = 0; i < teste.length; i++) {
            teste[i].join();
        }

    }
}`package examesFSO.exame2020_normal;

import java.util.concurrent.Semaphore;

public class Barrier {
    private int capacity;
    private Semaphore s, exclusao, counter;

    public Barrier(int capacity) {
        this.capacity = capacity;
        counter = new Semaphore(0);
        s = new Semaphore(0);
        exclusao = new Semaphore(1);
    }

    public void espera() throws InterruptedException {
        exclusao.acquire();
        if (counter.availablePermits() < capacity - 1) {
            counter.release();
            exclusao.release();

            s.acquire();
        } else {
            System.out.println("RELEASE ALL");
            for (int i = 0; i < capacity; i++) {
                s.release();
            }
        }
        exclusao.release();
    }

}

class TesteThread extends Thread {
    private Barrier b;
    private long waitPeriod;

    public TesteThread(long wait, Barrier b) {
        this.b = b;
        this.waitPeriod = wait;
        System.out.println("Thread instanciada " + this.getName());

    }

    public void espera() throws InterruptedException {
        b.espera();
    }

    @Override
    public void run() {
        try {
            System.out.println("Thread a dormir " + this.getName());
            sleep(waitPeriod);
            System.out.println("Thread a esperar " + this.getName());
            espera();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

class BarrierExample {

    public static void main(String[] args) throws InterruptedException {
        Barrier BR = new Barrier(5);

        TesteThread[] teste = new TesteThread[5];
        for (int i = 0; i < teste.length; i++) {
            teste[i] = new TesteThread((long) (Math.random() * 1000), BR);
            teste[i].start();
        }
        for (int i = 0; i < teste.length; i++) {
            teste[i].join();
        }

    }
}