What Java concurrency classes can help me?

Question

I am trying to write code that has the following behavior:

• There are many simultaneous and random calls to X by various threads (Thread X)
• At some point in the future, one call is made to Y by one thread (Thread Y)
• Until Y is called, X must go through unchallenged, while X's simultaneous calls work fine
• After calling Y, any existing X calls should be terminated, but new X calls must be rejected in some way (RuntimeException, etc.)
• Y should not be executed until all existing calls to X are completed.
• Refresh . When Y is called, it should send a signal to objects that are running in Thread X, which tells them to abort () in an incorrect way (ideally ending very quickly so that Y can continue)

I looked at using Semaphore, CountDownLatch, and even recording my own AbstractQueuedSynchronizer, but none of them met the above requirements. For example, CountDownLatch assumes that you know how many calls to X will be made, which we do not know.

It seems like I almost need a mixture of CountUpDownLatch and maybe some kind of simple AtomicBoolean, but this is what I am using now, and from time to time I am at a dead end. Again, I use a sketchy-looking implementation of CountUpDownLatch, which is part of HSQLDB, which does not seriously look thread safe.

Any ideas on how I should approach this issue?

Answer 1

Looks like a task ReentrantReadWriteLock. Something like that:

class A {
    private final ReadWriteLock lock = new ReentrantReadWriteLock();

    public void x() {
        if (!lock.readLock().tryLock()) throw new RuntimeException();
        try {
            ...
        } finally {
            lock.readLock().unlock();
        }
    }

    public void y() {
        lock.writeLock().lock();
        try {
            ...
        } finally {
            lock.writeLock().unlock();
        }
    }
}

Answer 2

Java 7 concurrency, java.util.concurrent.Phaser. - , ( ) . JSR 166 Interest Site.

, volatile boolean yHasEnetered, false. , :

        if(yHasEnetered)
            throw new IllegalStateException();
        phaser.register();
        //do work here
        phasre.arrive();

, Y , .

        yHasEntered=true;
        int phase = phaser.register();
        phaser.arriveAndAwaitAdvance(phase);
        //do y work here
        yHasEntered=false;

Y. , , , , , .

Answer 3

ExecutorService. submit(), , shutdown() - , , .

Answer 4

, ActiveObject?

class X {
    private static final ExecutorService service = Executors.newCachedThreadPool();


    public void x(){
       //Using anonymous class for brevity. Normal Runnable is ok. May be static inner class.  
       service.submit(new Runnable(){
                             @Override
                             public void run(){
                                   //do stuff
                             }
                       });
    }

    //Derived from Java6 ExecutorService API        
    public void shutDownAndAwait(){
        service.shutdownNow();
        try{
           if (!service.awaitTermination(60, TimeUnit.SECONDS))
              throw new TerminationException("Service did not terminate");
           }
        } catch (InterruptedException ie) {
           // (Re-)Cancel if current thread also interrupted
           service.shutdownNow();
           // Preserve interrupt status
           Thread.currentThread().interrupt();
        }
   }//shutdownAndAwait

}//X


class Y {

    private final X x = //inject reference

    public void y(){
        x.shutdownAndAwait();
    }
}

, x() , , x(), , . y()...