CODEWITHSUNDEEP
javamultithreading
Bipod
Bipod

Reputation: 69

Beginner Java Multithreading issue

I have a problem with a Java multithreaded file crawler that I am making. My issue is that I have a workQueue that is a linkedBlockingQueue that contains the names of files that I would like to crawl over with my threads, each thread will take() from the workQueue and while scanning through the file it may put() another file name into the workQueue (It is a dependency checker program). So I am never really certain when the work is all finished and all threads will eventually enter a waiting state when they try to take() from the (eventually) empty workQueue.

So I guess my question is, is there an efficient way to terminate all of threads once all of the work is finished (when all of threads have entered a waiting state)? Currently I just use sleep() on the main thread and then interrupt() all of the worker threads.

Sorry if the question sounds confused.

Upvotes: 2

Views: 106

Answers (3)

David Ehrmann
David Ehrmann

Reputation: 7576

There's a pattern called the Poison Pill that's good for this. Basically, when the producers are done, insert a special value into the queue that tells a consumer to stop. You can either insert one pill for each consumer, or, once a consumer gets a poison pill, return it to the queue for the next consumer. Since it sounds like you're just enqueuing strings, something like

public static final String POISON_PILL = "DONE";

Or in Java 8, use Optional to wrap your values, then have not present be the pill.

BlockingQueue<Optional<...>> queue;

Another option is using an ExecutorService (which is actually backed by a BlockingQueue) and submitting each file as its own task, then using executorService.shutdown() when you're done. The problem with this is that it couples your code more tightly than needed, and it makes it harder to reuse resources like database and HTTP connections.

I'd avoid interrupting your workers to signal them because that can cause blocking IO operations to fail.

Upvotes: 1

Guillaume F.
Guillaume F.

Reputation: 6463

I've had this problem before, and the only way I found was to send a special marker object to the BlockingQueue. When the Queue .take() the object, if this is the marker, then the Thread ends itself.

I've tried other solutions, like to wake up the thread and detect the Exception, with no success.

Upvotes: 1

JayC667
JayC667

Reputation: 2576

You might use the approch below. Add observer pattern if you need to. Or simply - instead of signalling with a death packet, collect a list of waiting Threads and then interrupt() them.

public class AccessCountingLinkedPrioQueue<T> {

    private final LinkedBlockingQueue<T>    mWrappingQueue                  = new LinkedBlockingQueue<>();
    private final Object                    mSyncLockObj                    = new Object();

    private final int                       mMaxBlockingThreads;
    private final T                         mDeathSignallingObject;

    private volatile int                    mNumberOfThreadsInAccessLoop    = 0;

    public AccessCountingLinkedPrioQueue(final int pMaxBlockingThreads, final T pDeathSignallingObject) {
        mMaxBlockingThreads = pMaxBlockingThreads;
        mDeathSignallingObject = pDeathSignallingObject;
    }

    public T take() throws InterruptedException {
        final T retVal;
        synchronized (mSyncLockObj) {
            ++mNumberOfThreadsInAccessLoop;
        }
        synchronized (mWrappingQueue) {
            if (mNumberOfThreadsInAccessLoop >= mMaxBlockingThreads && mWrappingQueue.isEmpty()) signalDeath();
            retVal = mWrappingQueue.take();
        }
        synchronized (mSyncLockObj) {
            --mNumberOfThreadsInAccessLoop;
        }
        return retVal;
    }

    private void signalDeath() {
        for (int i = 0; i < mMaxBlockingThreads; i++) {
            mWrappingQueue.add(mDeathSignallingObject);
        }
    }

    public int getNumberOfThreadsInAccessLoop() {
        return mNumberOfThreadsInAccessLoop;
    }
}

class WorkPacket {
    // ... your content here
}

class MultiThreadingBoss {
    static public final WorkPacket  DEATH_FROM_ABOVE    = new WorkPacket();

    public MultiThreadingBoss() {
        final int THREADS = 7;
        final AccessCountingLinkedPrioQueue<WorkPacket> prioQ = new AccessCountingLinkedPrioQueue<>(THREADS, DEATH_FROM_ABOVE);
        for (int i = 0; i < THREADS; i++) {
            final ThreadedWorker w = new ThreadedWorker(prioQ);
            new Thread(w).start();
        }
    }
}

class ThreadedWorker implements Runnable {
    private final AccessCountingLinkedPrioQueue<WorkPacket> mPrioQ;

    public ThreadedWorker(final AccessCountingLinkedPrioQueue<WorkPacket> pPrioQ) {
        mPrioQ = pPrioQ;
    }

    @Override public void run() {
        while (true) {
            try {
                final WorkPacket p = mPrioQ.take();
                if (p == MultiThreadingBoss.DEATH_FROM_ABOVE) break; // or return

                // ... do your normal work here

            } catch (final InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

Upvotes: 0

Related Questions