prevent duplicate queueing in the prio semaphore

[zpuller]

Follow up to #11376 incorporating a fix to prevent duplicate queueing in the event of spurious wakeups

[jlowe]

I didn't understand the last part exactly

Sorry, I should have been more clear. The idea is to add a boolean to ThreadInfo that is constructed as false but is set to true by the thread that is going to signal the wakeup, that way the thread waking up can distinguish between a spurious wakeup and an explicit one. A spurious wakeup should not leave the while loop, even if there are permits, because that can cause a priority inversion. Consider the case where a spurious wakeup on a low priority thread occurs at the same time the high priority thread is signaled. If the low priority thread wakeup races ahead of the high priority signal, it can end up grabbing the semaphore ahead of the high priority thread which is incorrect.

So the logic would look something like this:

private case class ThreadInfo(priority: T, condition: Condition, numPermits: Int) {
  var signaled: Boolean = false
}

def acquire(numPermits: Int, priority: T): Unit = {
  lock.lock()
  try {
    if (canAcquire(numPermits)) {
      commitAcquire(numPermits)
    } else {
      val info = ThreadInfo(priority, condition, numPermits)
      waitingQueue.enqueue(info)
      while (!info.signaled) {
        info.condition.await()
      }
    }
  } finally {
    lock.unlock()
  }
}

def release(numPermits: Int): Unit = {
  lock.lock()
  try {
    occupiedSlots -= numPermits
    // acquire and wakeup for all threads that now have enough permits
    var done = false
    while (!done && waitingQueue.nonEmpty) {
      val nextThread = waitingQueue.head
      if (canAcquire(nextThread.numPermits)) {
        val popped = waitingQueue.dequeue()
        assert(popped eq nextThread)
        commitAcquire(nextThread.numPermits)
        nextThread.signaled = true
        nextThread.condition.signal()
      } else {
        done = true
      }
    }
  } finally {
    lock.unlock()
  }
}

[revans2]

@jlowe in your code if there is an IterruptedException thrown when the acquire is waiting on the condition, then we need to make sure we are removed from the queue or the assumption that everything in the queue has a thread that is waiting on the condition is not true.

[jlowe]

if there is an IterruptedException thrown when the acquire is waiting on the condition, then we need to make sure we are removed from the queue

Nice catch! That's why I said, "it would look something like this." 😆

However it is important to note that even in the original code, your comment applies. If we leave acquire exceptionally, we need to pull ourselves out of the queue since we're no longer waiting to acquire.

[jlowe]

Thanks for all the work on this, @zpuller!

[jlowe]

build

[zpuller]

Thank you for the meticulous review!