Why can race condition in LogWriter cause the producer to block? [Concurrency in practice]

Question

First of all to prevent mark question as duplicate by guys who don't like read to the end I have read Producer-Consumer Logging service with Unreliable way to shutdown question. But it is not fully answer the question and answer contradicts the book text.

In book provided following code:

public class LogWriter {
    private final BlockingQueue<String> queue;
    private final LoggerThread logger;
    private static final int CAPACITY = 1000;

    public LogWriter(Writer writer) {
        this.queue = new LinkedBlockingQueue<String>(CAPACITY);
        this.logger = new LoggerThread(writer);
    }

    public void start() {
        logger.start();
    }

    public void log(String msg) throws InterruptedException {
        queue.put(msg);
    }

    private class LoggerThread extends Thread {
        private final PrintWriter writer;

        public LoggerThread(Writer writer) {
            this.writer = new PrintWriter(writer, true); // autoflush
        }

        public void run() {
            try {
                while (true)
                    writer.println(queue.take());
            } catch (InterruptedException ignored) {
            } finally {
                writer.close();
            }
        }
    }
}

Now we should to understand how to stop this process. We should stop logging but should not skip already commited messages.

Author researches approach:

public void log(String msg) throws InterruptedException {
     if(!shutdownRequested)
           queue.put(msg);
     else
           throw new IllegalArgumentException("logger is shut down");
 }

and comment it like this:

Another approach to shutting down LogWriter would be to set a “shutdown requested” flag to prevent further messages from being submitted, as shown in Listing 7.14. The con- sumer could then drain the queue upon being notified that shutdown has been requested, writing out any pending messages and unblocking any producers blocked in log . However, this approach has race conditions that make it unreliable. The implementation of log is a check-then-act sequence: producers could observe that the service has not yet been shut down but still queue messages after the shutdown, again with the risk that the producer might get blocked in log and never become unblocked. There are tricks that reduce the likelihood of this (like having the consumer wait several seconds before declaring the queue drained), but these do not change the fundamental problem, merely the likelihood that it will cause a fail- ure.

The phrase build enough hard for me.

I understand that

 if(!shutdownRequested)
           queue.put(msg);

is not atomic and message can be added to the queue after shutdowning. Yes, it is not very accurate but I don't see problem. queue just will be drain and when queue will be empty we can stop LoggerThread. Especially I don't understand why producers can be blocked.

Author didn't provide full code thus I cannot understand all details. I believe that this book was read by community majority and this example has detailed explanation.

Please explain with full code example.

Answer 1

The first thing to understand is that when a shutdown is requested, the producer needs to stop accepting any more requests and the consumer (LoggerThread in this case) needs to drain the queue. The code you present in the question demonstrates only one side of the story; the producer rejecting any further requests when shutdownRequested is true. After this example, the author proceeds to say :

The consumer could then drain the queue upon being notified that shutdown has been requested, writing out any pending messages and unblocking any producers blocked in log

First and foremost, queue.take in LoggerThread as shown in your question will block infinitely for new messages to be available in the queue; however, if we want to shutdown the LoggerThread(gracefully), we need to make sure that the shutdown code in LoggerThread gets a chance to execute when shutdownRequested is true rather than infinitely being blocked by queue.take.

When the author says that the consumer can drain the queue, what he means is that the LogWritter can check shutdownRequested and if it is true, it can call the non blocking drainTo method to drain the current contents of the queue in a separate collection instead of calling queue.take (or call a similar non blocking method instead). Alternatey, if shutdownRequested is false, LogWriter can proceed to call queue.take as usual.

The real problem with this approach is the way the log method (being called by producers) is implemented. Since it is not atomic, it is possible that multiple threads can miss the setting of shutdownRequested to true. What happens if the number of threads that miss this update is greater than the CAPACITY of the queue. Let's take a look at the log method again. (Added curly braces for the sake of explanation) :

public void log(String msg) throws InterruptedException {
     if(!shutdownRequested) {//A. 1001 threads see shutdownRequested as false and pass the if condition.

           //B. At this point, shutdownRequested is set to true by client code
           //C. Meanwhile, the LoggerThread which is the consumer sees that shutdownRequested is true and calls 
           //queue.drainTo to drain all existing messages in the queue instead of `queue.take`. 
           //D. Producers insert the new message into the queue.    
           queue.put(msg);//Step E
     } else
           throw new IllegalArgumentException("logger is shut down");
     }
}

As shown in Step E, it is possible for multiple producer threads to call put while the LoggerThread finished draining the queue and exited the w. There should be no issues till the 1000th thread calls put. The real issue is when the 1001th thread calls put. It will block as the queue capacity is only 1000 and the LoggerThread may no longer be alive or subscribed to the queue.take method.