5
\$\begingroup\$

I have implemented several lock-free algorithms to maximise throughput but have always baulked when performing a spin-lock. I can usually convince myself that this is the only way but it still nags at me that a spin-lock is a terribly bad idea.

I recently came across a FIFOMutex code fragment and thought perhaps that would be a solution. I have hacked it around a bit because I don't want mine to consume interrupt exceptions but essentially it looks like this:

/**
 * Queue of blocked writers.
 */
private static class BlockedQueue {
  // Everyone who is waiting.
  private final Queue<Thread> waiting = new ConcurrentLinkedQueue<>();
  // Ensure only one thread unlocks at a time.
  private final AtomicBoolean lock = new AtomicBoolean(false);

  public void sleep() {
    Thread me = Thread.currentThread();
    // Put me in the queue.
    waiting.add(me);
    try {
      do {
        // Park until I am head of the queue and I hold the lock
        LockSupport.park(this);
        /*
         * If I am not at the head of the queue - go back to sleep.
         *
         * If I am at the head of the queue - grab the lock and get out of the loop.
         *
         * If I was at the head and I couldn't get the lock - go back to sleep -
         * someone else is still in the process of exiting.
         */
      } while (!(waiting.peek() == me && lock.compareAndSet(false, true)));
    } finally {
      // Take me from the queue - it must be me because I locked everyone else out.
      waiting.remove();
      // Done with the lock.
      lock.set(false);
    }
  }

  public void wakeup() {
    // Wake the first in the queue ... unpark does nothing with a null parameter.
    LockSupport.unpark(waiting.peek());
  }
}
// Use this to block for a while.
private final BlockedQueue block = new BlockedQueue();

/**
 * Wait for a time when it is likely that a free slot is available.
 */
private void waitForFree() {
  // Still full?
  while (isFull()) {
    // Park me 'till something is removed.
    block.sleep();
  }
}

/**
 * A slot has been freed up. If anyone is waiting, let the next one know.
 */
private void signalFree() {
  block.wakeup();
}

See how I can now call block.sleep() in what used to be a spin-lock around while(isFull()).

What I like most about this is that the wakeup method is truly simple and that it will only wake up the thread that has been waiting longest.

I have some questions though:

  1. Will this do what I want it to do - i.e. handle multiple threads waiting for a resource without consuming unnecessary CPU.
  2. Is the while ((removed = waiting.remove()) != me) necessary? Logic suggests that the head of the queue at that point must be me but is there a scenario where it may not be?
\$\endgroup\$
6
  • \$\begingroup\$ Note - I have tinkered with it a bit since first post so if you have taken a copy to test, please recopy. \$\endgroup\$ Nov 27, 2013 at 14:34
  • \$\begingroup\$ also I am adding an answer ... just takes some time. \$\endgroup\$
    – rolfl
    Nov 27, 2013 at 14:36
  • \$\begingroup\$ Tweaked again - please take a new copy. \$\endgroup\$ Nov 27, 2013 at 15:26
  • \$\begingroup\$ I think the problem with the current/latest revision is that the first thread in to your system will need a wakeup() to work \$\endgroup\$
    – rolfl
    Nov 27, 2013 at 15:30
  • \$\begingroup\$ @rolfl - that was the intent because of private void waitForFree() { while (isFull()) { block.sleep(); } }. It will only sleep if the queue was full at one point. Note that LockSupport.unpark ... Otherwise, its next call to park is guaranteed not to block.. Or am I wrong? \$\endgroup\$ Nov 27, 2013 at 15:39

2 Answers 2

2
\$\begingroup\$

Posting a second answer to keep things clean.

I think I missed one of the important concepts in your use-case, that you are 'gating' the access to the resource from the resource-side of things, not the thread side of things....

If I were to suggest the following code, I think it would add to this 'conversation':

import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;


public class ResourceWaitQueue {

    private final ReentrantLock lock = new ReentrantLock(true); // use a FIFO lock.
    private final Condition goodtogo = lock.newCondition();
    private int releasecount = 0; // how many threads should be released.

    public final void await() throws InterruptedException {
        lock.lock();
        try {
            while (releasecount == 0) {
                goodtogo.await();
            }
            // releasecount > 0 and we were signalled.
            if (--releasecount > 0) {
                // reduce the releasdecount, but there's still another
                // thread that should be released.
                goodtogo.signal();
                // when that other thread releases, if there's still more
                // to be released, it can do that for us.
            };
        } finally {
            lock.unlock();
        }
    }

    public final void release() {
        lock.lock();
        try {
            // indicate there is work to do
            releasecount++;
            // signal the condition is true.... (only signal, not signalAll())
            goodtogo.signal();
        } finally {
            lock.unlock();
        }
    }


}

The above will queue threads up, using the 'fair' (FIFO) nature of the Re-Entrant Lock (or you can use the faster 'unfair' version too). It uses the lock to queue up threads, and it will release as many threads as times you call release.

\$\endgroup\$
2
  • \$\begingroup\$ You are right - I should have gone for a Condition in the first place and this is certainly the correct approach. It just seems strange to write a lock-free high-performace mechanism and then make use of locks and blocks when really all I want is a reliable and lightweight Thread.yield(). \$\endgroup\$ Nov 28, 2013 at 9:02
  • \$\begingroup\$ @OldCurmudgeon: I may have missed something but what about a fair Semaphore? Wouldn't it good? \$\endgroup\$
    – palacsint
    Mar 9, 2014 at 8:34
2
\$\begingroup\$

I believe there's race conditions in your code, Specifically, if you call wakeup() asymettrically (more times than you call sleep()) then you run the risk of a few things:

  1. you could have multiple running threads (i.e. there's not just one thread using the resource)
  2. you may possibly trigger a condition where calling wakeup() twice, very fast, will cause the same thread to be unparked (i.e. waiting.peek() in wakeup()) will return the same value twice....).
  3. it is possible that the queue does not contain a value when the wakeup is called because the 'active' thread is in the process of looking for itself (and removing other content from the queue) as it goes.

I have tried to find a way to make it work to my satisfaction below, but, frankly, I don't think it is possible without changing the code so much it is no longer 'yours'.

I think the reality is that a secure mechanism for this will necessarily rely on having a lock that contains both the unpark and the queue management.... at which point, using the park/unpark is no longer the right tool, so it defeats this process...

Feel free to read through what I was about to write, but, I am essentially abandoning it....


DISREGARD BELOW HERE

Right, you'll have to work with me on this one.... (it takes multiple 'heads' to work around 'threads'....)...

First, let me restate the problem I think you are trying to solve:

  • you have a resource that only one thread can access at a time
  • you want the 'waiting' threads to be queued (essentially in FIFO order) until the resource is available.
  • you want the waiting threads to be (b)locked rather than spinning.

You want the 'outside' code to look something like (this code will be in the resource-using threads):

try {
    blockedqueue.sleep(); // wait for my turn to use the resource
    doSomethingWith(resource);
} finally {
    blockedqueue.wakeup(); // indicate that another thread can use the resource.
}

I see that you start the BlockedQueue off in an 'unblocked' state, so the first thread in will immediately return. Only 'subsequent' threads will wait.

There is a gap in your logic though.... (hate to be the bearer of bad news....)...

It is possible for the queue to be empty when wakeup() is called even though there are blocked threads. This is because you may be removing() items in your loop at the same time as when the wake-up is called.

Also, I think that you should be more precise about what threads can re-awaken the queue. I like the try-finally concept which will assure that the same thread that calls sleep() will also call wakeup().

Consider the following alteration. Instead of using an AtomicBoolean to lock the thread, use an AtomicReference (and I have renamed it to 'active' instead of 'lock'....

public void wakeup() {
    // Unlock.
    if (active.compareAndSet(this, null)) {
        // Wake the first in the queue (and remove it too) ... unpark does nothing with a null parameter.
        LockSupport.unpark(waiting.poll());
    } else {
        throw new IllegalMonitorStateException("Cannot unlock a thread unless you are the owner of the lock");
    }
}

This will prevent spurious wake-ups, but if you have asymmetrical code (more wakeup() than sleep()) you will get exceptions (I think that's a good thing).

In your sleep() method you can also gain some efficiency by not using the queue at all if you're the only active thread...

public void sleep() {
    final Thread me = Thread.currentThread();
    if (blocked.compareAndSet(null, me) {
        // nothing is waiting, next threads will, but we can just keep going
        // there is a slight possibility that we can 'jump the queue'
        // if sleep() is called at **just** the right time...
        return;
    }

    // here we know that some other thread is active....
    // Put me in the queue.
    waiting.add(me);
    try {
        // Block while not first in queue or we're blocked
        // wait till we are at the front of the queue and we are also
        // not blocked by anything. If we are blocked, then park.
        // If not, we immediately block the next front-of-queue as well.
        while (waiting.peek() != me || !blocked.compareAndSet(false, true)) {
            LockSupport.park(this);
        }
    } finally {
       // Take me from the queue.
       Thread removed;
       while ((removed = waiting.remove()) != me) {
           // Put it back! It wasn't me!
           waiting.add(removed);
       }
    }
}

I think you may have a problem with initial state, and spurious wake-ups.... Consider the following

You have multiple instances of a Runnable (ta and tb for thread A and thread B), each of them does blockingqueue.sleep(). The expectation is that they will be woken up in turn when the queue's wakup() method is called (wake up both ta and tb by calling wakeup() twice).

So, both threads are asleep, the queue contains [ta, tb] and the atomic boolean blocked is false (the

\$\endgroup\$
4
  • \$\begingroup\$ Thanks for the attention. I think I picked up on your point about the initial condition problem and tweaked the code to deal with that - see the do { ... } while ... rather than the while ... {...}. Still thinking about your other points ... \$\endgroup\$ Nov 27, 2013 at 15:07
  • \$\begingroup\$ I know my answer is a mess... and any threading problem is hard to describe. If you have any questions, feel free to discuss. \$\endgroup\$
    – rolfl
    Nov 27, 2013 at 15:10
  • \$\begingroup\$ Of your points 1, 2 and 3 - I think the while ((removed = ... loop can be replaced completely by just waiting.remove() like in the original code because the logic dictates that the head of the queue must be me at this point. If that is the case then I suspect your concerns are unfounded. I think there is an issue in that area but I cannot define it at this time and I am still not certain you are wrong. \$\endgroup\$ Nov 27, 2013 at 15:13
  • \$\begingroup\$ I've tweaked it again and added some more comments - mostly to clarify my own thoughts. It seems to still work but I haven't managed to work out an effective test for it yet. \$\endgroup\$ Nov 27, 2013 at 15:28

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.