Java Thread Race Condition with BlockingQueue

Question

I have a potential race condition in my Java code that uses a BlockingQueue, and I'd like to know how to modify the code to avoid it:

private static BlockingQueue<FileToFTP> ftpQueue = new LinkedBlockingQueue<FileToFTP>();

private static boolean shuttingDown = false;

private static FileToFTP pendingFile = null;
private static int uploadsPending = 0;

private static Thread ftpThread = new Thread(new Runnable() { 
            public void run() {
                try {
                    for(;;) {
                        FileToFTP f2f = ftpQueue.take();    // this blocks until there is something to take
                        // FIXME: If the main thread takes over right 
                        // at this instant, then uploadsRemaining() 
                        // is not correct!!
                        uploadsPending = 1;
                        int qs = ftpQueue.size();
                        logIt(qs, f2f);
                        pendingFile = f2f;
                        if(!doUploadFile(f2f.getPath(), f2f.getFilename(), f2f.getRenameTo(), f2f.isBinary())) {
                            if(shuttingDown) {
                                log.info("Upload " + f2f + " failed!");
                            } else {
                                ftpQueue.offer(f2f);    // put it back on to retry later
                            }
                            uploadsPending = 0;
                        } else {
                            pendingFile = null;
                            uploadsPending = 0;
                        }
                        if(qs == 0) logIt("");
                    }
                } catch (InterruptedException consumed) {
                    // Allow thread to exit
                }
            }
        });

public static int uploadsRemaining() {
    return ftpQueue.size() + uploadsPending;
}

Please see the "FIXME" comment in the code. Thanks!!

Answer 1

Maybe I'm misinterpreting what you want, but it sounds like you may be better off using an ExecutorService to actually run things. You can create those using Exectors.newSingleThreadExecutor() or Executors.newFixedThreadPool(2) (2 being an example of the number of threads for it to use).

You can then either .execute or .submit Runnables on the ExecutorService. submit will return a Future<T> object that can be used to track the status of a particular job submitted to an ExecutorService.

Having said that, you may need to create a new class to do it as the Runnable / Callable class would need to have a FileToFTP variable in it. (Side note: I try to avoid inner classes, so if you're wondering why I didn't suggest doing it that way...)

The problem then becomes "how do I tell how many files are pending?" Unfortunately, the only easy way is to make it a static property of another class... with either static get/set methods or as a public/protected property. An AtomicInteger is ideal for this, so you may want to consider using one as a protected static one in your calling class. It has dedicated increment/decrement commands as well as ones that both adjust the value and return it.