Как написать простой честный семафор?

Question

Я нашел простую реализацию семафора (мой CustomSemaphore), и, насколько я понимаю, это «несправедливо», потому что в защищенный блок все время может входить только первый поток (не уверен).Как написать честный семафор (аналог параллелизма new Semaphore(1, true);)

   public class SimpleSemaphoreSample2 {
    CustomSemaphore cSem = new CustomSemaphore(1);

    public static void main(String[] args) {
        SimpleSemaphoreSample2 main = new SimpleSemaphoreSample2();
        Semaphore sem = new Semaphore(1, true);
        Thread thrdA = new Thread(main.new SyncOutput(sem, "Thread1"), "Thread1");
        Thread thrdB = new Thread(main.new SyncOutput(sem, "Thread2"), "Thread2");

        thrdA.start();
        thrdB.start();

        try {
            thrdB.join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("END");
    }

    class SyncOutput implements Runnable {
        private Semaphore sem;
        private String msg;

        public SyncOutput(Semaphore s, String m) {
            sem = s;
            msg = m;
        }

        @Override
        public void run() {
            while (true) {
                try {
//                  sem.acquire();
                    cSem.acquire();
                    System.out.println("Before");
                    Thread.sleep(500);
                    System.out.println(msg);
                    Thread.sleep(500);
                    System.out.println("After");
                    Thread.sleep(500);
                } catch (Exception exc) {
                    exc.printStackTrace();
                }
//              sem.release();
                cSem.release();
            }
        }
    }

    public class CustomSemaphore {
        private int counter;

        public CustomSemaphore() {
            this(0);
        }

        public CustomSemaphore(int i) {
            if (i < 0)
                throw new IllegalArgumentException(i + " < 0");
            counter = i;
        }

        public synchronized void release() {
            if (counter == 0) {
                this.notify();
            }
            counter++;
        }

        public synchronized void acquire() throws InterruptedException {
            while (counter == 0) {
                this.wait();
            }
            counter--;
        }
    }
}
enter code here

Answer 1

Ваш семафор нечестен, потому что возможно, что поток ждет вечно. Подумайте о мьютексе (двоичном семафоре), который используется для записи значения тремя потоками. T1 получают, T2 ждут и T3 ждут. Теперь во время выпуска вы уведомляете, и один между T2 и T3 берет семафор (скажем, T2). Теперь Т1 вернись и подожди. Когда T2 уведомляет, T1 принимает его. Это может происходить столько раз, сколько возможно, и у T3 никогда не будет семафора.

Одним из изменений может быть использование простого FIFO внутри семафора. Когда поток должен ждать, вы добавляете его идентификатор в очередь. Теперь, когда вы уведомляете, вы уведомляете все темы. Единственный поток, чей прогресс - это тот, который находится в начале очереди.

Answer 2

Согласно Параллелизму Java на практике говорится, что

Внутренняя блокировка не дает никаких детерминированных гарантий справедливости

Внутренняя блокировка здесь используется synchronized. Таким образом, вы не сможете сделать этот пример семафора справедливым, не заменив synchronized на Lock lock = new ReentrantLock(true);

Где аргумент true как конструктор указывает ReentrantLock на fair

Редактировать на основе комментария @ trutheality

Если вы действительно хотите, чтобы это было правильно без использования ReentrantLock, вы можете реализовать Семафор, наследующий примитивы синхронизации от AbstractQueuedSynchronizer . Это может оказаться довольно сложным, и если вы можете правильно написать это с помощью ReentrantLock, я бы предложил это. Примечание: ReentrantLock делегирует свою синхронизацию AQS.

Answer 3

У меня есть пример повторяющегося семафора, но он предназначен только для 2 претендентов. Если вы хотите расширить код более чем на 2, вы должны реализовать простой список и внести несколько изменений, включая тест для wait() в методе aquire().

package nmscd.utils;

/**
 * A simple, non-reentrant, one permit, FAIR semaphore
 * @author cosmo
 */
public class SimpleSemaphore {

    private boolean aquired = false;
    private Thread currThread;
    private Thread releasedThread;
    private int pretendersCount = 0;

    public synchronized void aquire() throws InterruptedException {
        while ((Thread.currentThread() != currThread && aquired) || (pretendersCount > 0 && Thread.currentThread() == releasedThread)) {
            pretendersCount++;
            try {
                wait();
            } finally {
                pretendersCount--;
            }
        }
        aquired = true;
        currThread = Thread.currentThread();
    }

    public synchronized void release() {
        if (Thread.currentThread() == currThread) {
            aquired = false;
            currThread = null;
            releasedThread = Thread.currentThread();
            notifyAll();
        }
    }

}

Ключ в этом классе заключается в том, чтобы проверить в методе aquire, чтобы увидеть, является ли запрашиваемый поток нужным потоком, все остальные потоки должны ждать. Так что, если у вас достаточно информации, чтобы определить этот поток, вы можете выбрать, какой поток возвращает из aquire()