Simplified LogService (from concurrency in practice) with shutdown feature

Question

Brian Goetz provided following code of LogService with shutdown feature:

public class LogService {
    private final BlockingQueue<String> queue;
    private final LoggerThread loggerThread;
    private final PrintWriter writer;
    @GuardedBy("this") private boolean isShutdown;
    @GuardedBy("this") private int reservations;

    public LogService(Writer writer) {
        this.queue = new LinkedBlockingQueue<String>();
        this.loggerThread = new LoggerThread();
        this.writer = new PrintWriter(writer);
    }

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

    public void stop() {
        synchronized (this) {
            isShutdown = true;
        }
        loggerThread.interrupt();
    }

    public void log(String msg) throws InterruptedException {
        synchronized (this) {
            if (isShutdown)
                throw new IllegalStateException(/*...*/);
            ++reservations;
        }
        queue.put(msg);
    }

    private class LoggerThread extends Thread {
        public void run() {
            try {
                while (true) {
                    try {
                        synchronized (LogService.this) {
                            if (isShutdown && reservations == 0)
                                break;
                        }
                        String msg = queue.take();
                        synchronized (LogService.this) {
                            --reservations;
                        }
                        writer.println(msg);
                    } catch (InterruptedException e) { /* retry */
                    }
                }
            } finally {
                writer.close();
            }
        }
    }
}

I think that I have simpler variant(I don't understand why we need reservations counter), please review its safety:

public class LogService {
    private final BlockingQueue<String> queue;
    private final LoggerThread loggerThread;
    private final PrintWriter writer;
    @GuardedBy("this")
    private boolean isShutdown;


    public LogService(Writer writer) {
        this.queue = new LinkedBlockingQueue<String>();
        this.loggerThread = new LoggerThread();
        this.writer = new PrintWriter(writer);
    }

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

    public void stop() {
        synchronized (this) {
            isShutdown = true;
        }
        loggerThread.interrupt();
    }

    public void log(String msg) throws InterruptedException {
        synchronized (this) {
            if (isShutdown)
                throw new IllegalStateException(/*...*/);
            queue.put(msg); // inside synchronized now
        }

    }

    private class LoggerThread extends Thread {
        public void run() {
            try {
                while (true) {
                    try {
                        synchronized (LogService.this) {
                            if (isShutdown && queue.isEmpty())
                                break;
                        }
                        String msg = queue.take();
                        writer.println(msg);
                    } catch (InterruptedException e) { /* retry */
                    }
                }
            } finally {
                writer.close();
            }
        }
    }
}

Answer 1

Concurrency issue

I think one critical part is found here:

synchronized (LogService.this) {
    if (isShutdown && queue.isEmpty()) break;
}
--> ... <--
String msg = queue.take();
--> ... <--
writer.println(msg);

You query isShutdown and queue.isEmpty() but you cannot ensure this state anymore if you leave the monitor and access the queue afterwards.

After that you also cannot ensure the output order (println) that is specified by the queue.

Both should be done in the monitor.

GuardedBy obsolete

Furthermore I do not think that the GuardedBy annoptation is neccessary as isShutdown is already only requested and modified within the monitor (synchronized(this)).

@GuardedBy("this")
private boolean isShutdown;

Writer

I do not know if it is intended to close the PrinterWriter after the queue is empty. This will cause a problem when restarting the LogThread. Closed PrintWriters may not work anymore which depends on the underlying Writer that interpretes the "close()"-operation:

PrintWriter pw = new PrintWriter(new FileOutputStream("hello.txt"));
pw.close();
pw.println("Hallo"); // will produce no output in file

try with resources

If you want to close resources properly you should use following construct:

try (PrintWriter writer = new PrintWriter(...)) {
    ...
} catch (...) {
    ...
}

Another point is that you should have a symmetry in creating and closing a resource. Your LogService is now responsible to close the Writer but it doesn't create it. Either you provide a factory for the resource so the LogService is responsible for creating and closing it OR the user of the LogService that passes in the Writer should be aware of creation and closing. Nevertheless the maximum proximity for those operations is the goal.

State handling

Currently you micro manage the state handling (running/stopped). You should consider implementing a real state pattern as a great concept to make clear when which method is allowed to be called.

You have some semantic things I find confusing. Logging is allowed only when the service is not stopped but you may always start and stop the service.

Thread interrupt

Your thread should not be stopped through the interrupt method. Try to reformulate your communication with the thread to stop without using such harsh mechanism that throws an exception.

Restarting

You do allow starting the loggerThread although this will cause an exception. I would expect either NOOP or a domain specific exception as the caller is not interested in a thread specific exception. Should the caller even know that a thead is working? Another possibility is to start over with a new thread. You already handle state so the obvious functionality for calling the start()-method on a stopped LogService is to start it again.

Personal opinion

As this is I would trust neither one nor the other implementation as they are riddled with semantic ambiguities, concurrency issues and micro management of state. I suggest to rethink the concept with real state handling and a proper monitor definition.

Code

State

The state pattern will cover three methods:

private interface State {
    void start();
    void log(String message);
    void stop();
}

LogService

The LogService will delegate all public calls to the state pattern. It provides a queue that will buffer log messages.

Initially the LogService is in the state "StateStopped".

You see that we changed from a blocking queue to a normal queue that is not thread-safe. But that is ok as the LogService serves as a monitor around it beside the other two variables loggerThread and state.

LogTarget may be access concurrently but this is not our responsibility AND we can only ensure the correct implementation by convention.

public class LogService implements LogTarget, LogSource {

    // states
    ...

    private State state;

    private final Queue<String> messageQueue;
    private LogProcessorThread loggerThread;
    private final LogTarget logTarget;


    public LogService(LogTarget logTarget) {
        this.messageQueue = new ArrayDeque<String>();
        this.logTarget = logTarget;
        this.state = new StateStopped();
    }

    public void start() {
        this.state.start();
    }

    public void stop() {
        this.state.stop();
    }

    public void log(String message) {
        this.state.log(message);
    }

    public String getNextQueuedMessage() {
        synchronized (this) {
            return this.messageQueue.poll();
        }
    }

}

StateStopped

As the state pattern has a context object (here LogService) this context object will be used as the monitor (synchronized(LogService.this)).

My implementation accepts log messages put it will not notify any consumer.

Calling stop() will be a NOOP. If your flavour is to throw an exception do that.

Calling start() will instantiate a new thread and start it. After that the internal state is changed to be "StateRunning".

private class StateStopped implements State {

    @Override
    public void start() {
        synchronized (LogService.this) {
            loggerThread = new LogProcessorThread(LogService.this, logTarget);
            loggerThread.start();
            state = new StateRunning();
        }
    }

    @Override
    public void log(String message) {
        synchronized (this) {
            messageQueue.add(message);
        }
    }

    @Override
    public void stop() {
        // ignore or throw IllegalStateException... your choice
    }

}

StateRunning

As the state pattern has a context object (here LogService) this context object will be used as the monitor (synchronized(LogService.this)).

Calling start() will lead to a NOOP. Here you have the choice to throw an exception as well.

Calling log() will accept new log messages but as the service knows that a thread is available (StateRunning) that may wait for new messages it is notified.

Calling stop() will safely shutdown the thread by communicating a stop request.

We will change the internal state of the LogService within the monitor and waiting until the thread terminates outside the monitor.

private class StateRunning implements State {

    @Override
    public void start() {
        // ignore or throw IllegalStateException... your choice
    }

    @Override
    public void log(String message) {
        synchronized (this) {
            messageQueue.add(message);
            loggerThread.awakeOnWait(); // this is more a callback or an observer pattern
        }
    }

    @Override
    public void stop() {
        Thread lastBreath = null;
        synchronized (LogService.this) {
            lastBreath = loggerThread;
            loggerThread.requestStop();
            loggerThread = null;
            state = new StateStopped();
        }
        try {
            lastBreath.join();
        } catch (InterruptedException e) {
        }
    }

}

LogProcessorThread

This thread while refactoring seems to become more and more an artefact that will ask for log messages from somewhere and passes these log message to somewhere.

Here again: The thread itself is the monitor over "shouldStop" and the critical operations source.getNextQueuedMessage() and target.log(message) that should not be intercepted.

public class LogProcessorThread extends Thread {

    private LogTarget target;
    private LogSource source;
    private boolean shouldStop = false;

    public LogProcessorThread(LogSource source, LogTarget target) {
        this.target = target;
        this.source = source;
    }

    public void run() {
        boolean expectedMoreMessages = true;
        while (expectedMoreMessages) {
            expectedMoreMessages = logNextMessage();
        }
    }

    public synchronized void requestStop() {
        this.shouldStop = true;
        this.notify();
    }

    public synchronized void awakeOnWait() {
        this.notify();
    }

    public synchronized boolean logNextMessage() {

        boolean shouldStop = this.shouldStop; // keep the state stable in this method invocation

        String message = source.getNextQueuedMessage();
        if (message != null) {
            target.log(message);
        }
        if (message == null && !shouldStop) {
            try {
                this.wait();
            } catch (InterruptedException e) {
                throw new RuntimeException("unexpected", e);
            }
        }
        boolean expectMoreMessages = message != null || !shouldStop;
        return expectMoreMessages;
    }

}

LogSource, LogTarget

These are not neccessary but they show the inherent abstraction the thread makes use of. So A LogService is of course a LogSource but it can also be a LogTarget.

public interface LogSource {
    String getNextQueuedMessage();
}

public interface LogTarget {
    public void log(String message);
}

Answer 2

I agree with @oopexpert that neither implementation looks particularly nice.

I can see a few ways to improve this:

• isShutdown should probably be made volatile; or if you really want to ensure consistent behavior, maybe consider using AtomicBoolean instead.
• LinkedBlockingQueue is thread safe, so moving the offer inside the synchronized block seems redundant and increases the amount of lock contention on this.
• If you use an AtomicBoolean, you wouldn't need a synchronized block at all for queueing new messages.
• Alternatively, you could consider using a ReadWrite lock and use the writelock only when printing messages. This has the advantage of allowing multiple threads to add messages at the same time with a read lock.
• The writer threads could drain the queue and print everything instead of merely taking one item. This could end up being more efficient.
• The lock should only be acquired if the queue is not empty. Now basically all the writer threads are essentially acquiring a lock on every iteration, when arguably they mostly would not need it unless the queue is not empty.