Cleaning a buffer before termination

Question

I'm writing a program similar to the producer-consumer problem. Here's my main code:

public class PipeProcessor {

private volatile boolean close = false;

Pipe pipe;
Output out;

public PipeProcessor(Pipe pipe)
{
 this.pipe = pipe;
}

public void run()
{
 while(!close)
 {
  out.output(pipe.get());
 }

 while(pipe.size() > 0)
  out.output(pipe.get());

    out.close();

}

public void close()
{
 close = true;
}
}

Pipe is a wrapper for an ArrayBlockingQueue and acts as a buffer. Output is a class which takes an element in the buffer and outputs it.

I want to make sure that the PipeProcessor terminates cleanly, i.e. when it is signaled to close, it cleans the buffer. Since the close() method is called by a shutdown hook, I'm making sure that the buffer is not being filled while the processor is closing, Is this the right way to do it? Thank you.

Answer 1

Not sure why you are trying to clean the pipe when you close, why don't you discard it and let the GC clean it up? All you need is the close and the first loop as far as I can see.

Answer 2

It looks like your code does what you want it to do. You could make your code easier to understand if you take a look at your naming, for instance the boolean "close" could be named "closing" or "shuttingDown" or inverse it to "running" which would result in more readable code imho.

The while loop in run() and the lines following it could be written as:

    while (running || pipe.size() > 0) {

        out.output(pipe.get());
    }

Answer 3

If you want to process all the elements in the pipeline before the process stops, I don't think I'd actually use a shutdown hook - I'd shut the pipeline down explicitly in the main code, and wait for it to complete before letting the main thread finish. I suggest you either change the close() method to block until the pipeline has finished, or add a separate method (e.g. waitForPipelineToEmpty()).

In this way you can make it all more controllable - in particular, it means you won't be trying to process things while other bits of the system are cleaning themselves up in shutdown hooks.

One alternative way of terminating a producer/consumer queue is to have a sentinel value which means "stop now". Then you just feed that into the end of the pipeline (and avoid adding any more "real" values) - and your processor just stops when it sees that item.

Answer 4

I'm concerned that out.close() won't necessarily get called. If Pipe.get() blocks like ArrayBlockingQueue.take() and it doesn't return a sentinel value when it detects closure, then calling close() on PipeProcessor after the Pipe is empty will have no effect because the while(!close) condition will not be evaluated again.

But maybe (1) the Pipe will always be closed first, (2) Pipe.get() does detect closure and (3) it returns some sentinel value like null which Output can handle. If this is the case, then your code looks good.