Запретный вариант ожидания/уведомления

Question

Java требует, чтобы нить владела монитором o перед вызовом o.wait() или o.notify(). Это общеизвестный факт. Однако существуют ли блокировки мьютексов, которые необходимы для любого такого механизма? Что, если был API, который обеспечил

compareAndWait 

и

setAndNotify 

вместо того, чтобы, сочетая действие CAS с планированием потоков/descheduling? Это имело бы некоторые преимущества:

• темы, связанные с входом в состояние ожидания, не будут препятствовать прогрессу в уведомлении потоков;

• им также не придется ждать друг от друга, прежде чем им разрешат проверить состояние ожидания;

• на уведомляющей стороне любое количество потоков производителей может продолжаться одновременно.

Есть ли фундаментальное, непреодолимое препятствие для предоставления такого API?

Answer 1

Больше мыслительного эксперимента, чем какой-либо настоящий рабочий код, но это, похоже, работает.

// My lock class. 
public static class Padlock<E extends Enum<E>> { 

    // Using Markable because I think I'm going to need it. 
    public final AtomicReference<E> value; 
    // Perhaps use a set to maintain all waiters. 
    Set<Thread> waiters = ConcurrentHashMap.newKeySet(); 

    public Padlock(E initialValue) { 
     this.value = new AtomicReference<>(initialValue); 
    } 

    /** 
    * Waits for the locks value to become the specified key value. 
    * 
    * @param waitFor - The desired key. 
    */ 
    public void compareAndWait(E waitFor) { 
     log("Wait for " + waitFor); 
     // Spin on the value. 
     while (value.get() != waitFor) { 
      log("Park waiting for " + waitFor); 
      // Remember me as waiting. 
      waiters.add(Thread.currentThread()); 
      // TODO: What do we do here?? 
      LockSupport.park(); 
      log("Awoke " + waitFor); 
     } 
    } 

    /** 
    * Sets the locks value to the key value. 
    * 
    * If this resulted in a change - notify all changers. 
    * 
    * @param shouldBe - What it should be now. 
    * @param makeIt - The new value to set. 
    */ 
    public void setAndNotify(E shouldBe, E makeIt) { 
     log("Set " + shouldBe + "->" + makeIt); 
     if (value.compareAndSet(shouldBe, makeIt)) { 
      log("Notify " + shouldBe + "->" + makeIt); 
      // It changed! Notify the waiters. 
      for (Thread t : waiters) { 
       // Perhaps 
       log("Unpark " + t.getName()); 
       LockSupport.unpark(t); 
      } 
     } 
    } 
} 

enum State { 

    Off, On; 
} 

private static final long TESTTIME = 30000; 
private static final long TICK = 100; 

private static final void log(String s) { 
    System.out.println(Thread.currentThread().getName() + ": " + s); 

} 

static class MutexTester implements Runnable { 

    final Padlock<State> lock; 

    public MutexTester(Padlock<State> lock) { 
     this.lock = lock; 
    } 

    @Override 
    public void run() { 
     Thread.currentThread().setName(this.getClass().getSimpleName()); 
     long wait = System.currentTimeMillis() + TESTTIME; 
     do { 
      // Wait for an On! 
      lock.compareAndWait(Test.State.On); 
      try { 
       log("Got it!"); 
       try { 
        Thread.sleep(TICK); 
       } catch (InterruptedException ex) { 
        log("Interrupted!"); 
       } 
      } finally { 
       // Release 
       lock.setAndNotify(Test.State.On, Test.State.Off); 
      } 
     } while (System.currentTimeMillis() < wait); 
     log("Done"); 
    } 
} 

static class RandomSwitcher implements Runnable { 

    final Padlock<State> lock; 
    final Random random = new Random(); 

    public RandomSwitcher(Padlock<State> lock) { 
     this.lock = lock; 
    } 

    @Override 
    public void run() { 
     Thread.currentThread().setName(this.getClass().getSimpleName()); 
     long wait = System.currentTimeMillis() + TESTTIME; 
     do { 
      // On! 
      lock.setAndNotify(Test.State.Off, Test.State.On); 
      log("On!"); 
      pause(); 
      lock.setAndNotify(Test.State.On, Test.State.Off); 
      log("Off!"); 
      pause(); 
     } while (System.currentTimeMillis() < wait); 
     log("Done"); 
    } 

    private void pause() { 
     try { 
      // Random wait. 
      Thread.sleep(TICK * random.nextInt(10)); 
     } catch (InterruptedException ex) { 
      System.out.println("Interrupted! " + Thread.currentThread().getName()); 
     } 
    } 
} 

public void test() throws InterruptedException { 
    final Padlock<State> lock = new Padlock<>(State.Off); 
    Thread t1 = new Thread(new MutexTester(lock)); 
    t1.start(); 
    Thread t2 = new Thread(new RandomSwitcher(lock)); 
    t2.start(); 
    t1.join(); 
    t2.join(); 
} 

Я реализовал протокол, который compareAndWait ожидает исключительного использования в то время как setAndNotify освобождает мьютекс.

Answer 2

Во-первых, встроенные мониторы Java (synchronized и wait) более эффективны, чем многие могут подумать. См. biased locking, и планируется дальнейшее усовершенствование, использующее аппаратную транзакционную память.

Во-вторых, механизм, который вы ищете, и тот, который предоставляется synchronized/wait, служат для разностных целей. Последний защищает некоторый охраняемый ресурс и должен содержать блокировку, потому что предполагается, что после wait вы хотите находиться внутри критического раздела. То, что вы ищете, обеспечивается другими примитивами параллелизма Java, такими как CountDownLatch, Phaser или Semaphore.

Answer 3

Нет проблем с внедрением произвольных механизмов ожидания/уведомления с использованием LockSupport.park() и LockSupport.unpark(Thread), поскольку эти базовые примитивы не требуют фиксации каких-либо замков.

Причина, почему ни Object.wait/Object.notify ни Condition.await/Condition.signal предложение вам такое уведомление без проведения блокировки является семантическим один. Концепция уведомления заключается в том, что один поток ожидает выполнения условия, а другой останавливает ожидание, когда условие изменилось на выполненное состояние. Без фиксации, связанной с этим условием, нет гарантии, что условие не изменяется между тестами состояния состояния и изменением состояния потока.

Чтобы быть более конкретным, существует вероятность того, что когда поток, который изменил условие, уведомляет о другом потоке, это условие было изменено еще раз до того, как произойдет уведомление. Но еще хуже, условие может измениться на «выполнено» до того, как поток начнет wait, и в этом случае поток может пропустить уведомление и вешать навсегда.

Даже если вы можете слить тестовое условие и операцию ожидания в одну атомную операцию, это не поможет. Ожидание состояния не является самоцелью. Причина, по которой поток хочет ждать условия, состоит в том, что он хочет выполнить действие , условие которого является необходимым условием и, следовательно, не должно меняться во время выполнения действия. В этом весь смысл: тест условия и действие должны быть реализованы как одна операция с блокировкой, независимо от того, как реализована концепция блокировки.

Существуют особые случаи, когда такие проблемы не могут возникнуть, например. когда известно, что переходы состояний состояния ограничены, поэтому вы можете исключить, что условие может вернуться к невыполненному состоянию. Вот именно такие инструменты, как CountDownLatch, CyclicBarrier, Phaser, но механизм уведомления с предопределенной семантикой wait/notify подразумевает не предполагать такой особый случай.