0

I've found already good hints here. But I have much more difficult task - additional requests are:
- my low priority forever thread can be started/stopped from main thread (that's same) - but it also has to lock one resource for exclusive access.
- my low priority forever thread can be paused/continued from another high priority thread(s) (they then also lock and use that one resource)
- also I want the low priority thread to not release lock in each loop, but only release the lock once it is told to (for speed purposes - cos I would need to init/deinit resource) (If I would have release lock every loop then I may let java to manage competing threads and expect that high priority threads would win every now and then).

I come up with solution and I think it is safe and sound. But I'm newbie so I'm open for finding bug in my solution or suggestions for improvements. So my questions are:
1) Is my solution really thread safe in all cases?
2) Is it optimal solution or can I improve something?

If it is agreed by you then let it be used as a template.

Here is core code:

FlagNotify lowPriorRunReq   = new FlagNotify(false); // low priority task run request
FlagNotify lowPriorPauseReq = new FlagNotify(false); // low priority task pause request (it uses notify)
volatile boolean lowPriorRunFlag = false;  // low priority task run flag
Lock      groupLock        = new ReentrantLock();  // group lock (used to acquire lowPriorRunFlag always correctly)
Semaphore resourceSemaphore = new Semaphore(1);    // main semaphore protecting resource that has to be accessed sequentially

public class PrioritySingleTaskThread extends Thread {
    @Override
    public void run() {
        prn("High Priority Task created");
        groupLock.lock();
        if(lowPriorRunFlag == true) lowPriorPauseReq.setNotify(true);
        resourceSemaphore.acquireUninterruptibly();
        groupLock.unlock();
        accessResource("high priority task");
        resourceSemaphore.release();
        groupLock.lock();
        if(lowPriorPauseReq.get() == true) lowPriorPauseReq.setNotify(false);
        groupLock.unlock();
        prn("High Priority Task closed");
    }
}

public class LowPriorityContinuousThread extends Thread {
    void getResourceSemaphore(){
        groupLock.lock();
        resourceSemaphore.acquireUninterruptibly();
        lowPriorRunFlag = true;
        groupLock.unlock();
        accessResource("low priority init"); // here it is initialization and I want to happen only on request from priority thread
    }
    void releaseResourceSemaphore(){
        accessResource("low priority de-init"); // here it is de-initialization and I want to happen only on request from priority thread
        lowPriorRunFlag = false;
        resourceSemaphore.release();
    }

    @Override
    public void run() {
        while(true){
            //prn("Low Priority Run req: "+lowPriorRunReq.get());
            if(lowPriorRunReq.get() == true && lowPriorRunFlag == false){
                prn("Low  Priority Task starting");
                getResourceSemaphore();
                prn("Low  Priority Task started");
            }
            if(lowPriorRunReq.get() == false && lowPriorRunFlag == true){
                prn("Low  Priority Task stopping");
                releaseResourceSemaphore();
                prn("Low  Priority Task stopped");
                lowPriorRunReq.smartWait(true);
            }
            // note keep checking lowPriorRunFlag. Imagine there is RunFlag detected by high priority thread
            // before de-asserted, then pauseReq would be requested from high priority thread
            // then resource is released when low priority task stops.
            // High priority lock and use resource, but since pauseReq is set
            // this thread would try to access device in order to de-init and pause (unless we check runFlag)
            if(lowPriorPauseReq.get() == true && lowPriorRunFlag == true){
                prn("Low  Priority Task pausing");
                releaseResourceSemaphore();
                prn("Low  Priority Task paused");
                lowPriorPauseReq.smartWait(false);
                getResourceSemaphore();
                prn("Low  Priority Task continue");
            }
            if(lowPriorRunFlag){
                accessResource("low priority task");
            }
        }
    }
}

And here is full compilable Java code including test bench (so I have a hint it is a safe solution - but you never know with those threads)

import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class Main {
    // this inner class is only for setting flag and waiting to it by notify - which does not hog CPU
    public class FlagNotify{
        private Boolean flag; // do not synchro on Boolean - it is immutable, thus new object is created every value change....
        private Object synchro = new Object();

        public FlagNotify(boolean val) {
            flag = val;
        }

        public void setNotify(boolean val) {
            synchronized (synchro) {
                flag = val;
                synchro.notify();
            }
        }

        public boolean get(){
            return flag;
        }
        public void smartWait(boolean expVal){
            synchronized (synchro){
                while(flag != expVal){
                    try {
                        synchro.wait();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        }
    }

    FlagNotify lowPriorRunReq   = new FlagNotify(false); // low priority task run request
    FlagNotify lowPriorPauseReq = new FlagNotify(false); // low priority task pause request (it uses notify)
    volatile boolean lowPriorRunFlag = false;  // low priority task run flag
    Lock      groupLock        = new ReentrantLock();  // group lock (used to acquire lowPriorRunFlag always correctly)
    Semaphore resourceSemaphore = new Semaphore(1);    // main semaphore protecting resource that has to be accessed sequentially

    public class PrioritySingleTaskThread extends Thread {
        @Override
        public void run() {
            prn("High Priority Task created");
            groupLock.lock();
            if(lowPriorRunFlag == true) lowPriorPauseReq.setNotify(true);
            resourceSemaphore.acquireUninterruptibly();
            groupLock.unlock();
            accessResource("high priority task");
            resourceSemaphore.release();
            groupLock.lock();
            if(lowPriorPauseReq.get() == true) lowPriorPauseReq.setNotify(false);
            groupLock.unlock();
            prn("High Priority Task closed");
        }
    }

    public class LowPriorityContinuousThread extends Thread {
        void getResourceSemaphore(){
            groupLock.lock();
            resourceSemaphore.acquireUninterruptibly();
            lowPriorRunFlag = true;
            groupLock.unlock();
            accessResource("low priority init"); // here it is initialization and I want to happen only on request from priority thread
        }
        void releaseResourceSemaphore(){
            accessResource("low priority de-init"); // here it is de-initialization and I want to happen only on request from priority thread
            lowPriorRunFlag = false;
            resourceSemaphore.release();
        }

        @Override
        public void run() {
            while(true){
                //prn("Low Priority Run req: "+lowPriorRunReq.get());
                if(lowPriorRunReq.get() == true && lowPriorRunFlag == false){
                    prn("Low  Priority Task starting");
                    getResourceSemaphore();
                    prn("Low  Priority Task started");
                }
                if(lowPriorRunReq.get() == false && lowPriorRunFlag == true){
                    prn("Low  Priority Task stopping");
                    releaseResourceSemaphore();
                    prn("Low  Priority Task stopped");
                    lowPriorRunReq.smartWait(true);
                }
                // note keep checking lowPriorRunFlag. Imagine there is RunFlag detected by high priority thread
                // before de-asserted, then pauseReq would be requested from high priority thread
                // then resource is released when low priority task stops.
                // High priority lock and use resource, but since pauseReq is set
                // this thread would try to access device in order to de-init and pause (unless we check runFlag)
                if(lowPriorPauseReq.get() == true && lowPriorRunFlag == true){
                    prn("Low  Priority Task pausing");
                    releaseResourceSemaphore();
                    prn("Low  Priority Task paused");
                    lowPriorPauseReq.smartWait(false);
                    getResourceSemaphore();
                    prn("Low  Priority Task continue");
                }
                if(lowPriorRunFlag){
                    accessResource("low priority task");
                }
            }
        }
    }
//-------------------------------------------------------------------------
//-- following functions are meant only for testing
    AtomicInteger clashDetector = new AtomicInteger(0); // only for testing purposes

    public void accessResource(String from){
        prn("Resource used from "+from);
        if(clashDetector.addAndGet(1)>1){
            System.out.println("Clash detected - you are a bad programmer :((((((");
            System.exit(-1);
        }
        sleepRandom(5);
        clashDetector.getAndAdd(-1);
    }

    public void sleepRandom(long maxMiliSec){
        mySleep((long)(Math.random()*maxMiliSec));
    }

    public void mySleep(long miliSec){
        try{
            Thread.sleep(miliSec);
        }catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    void prn(String s){
        System.out.println(s);
    }

    public void test(){
        new LowPriorityContinuousThread().start();
        for(long i=0; i< (long)1e3; i++){
            lowPriorRunReq.setNotify(true);
            for(int j=0; j<Math.random()*100;j++){
                sleepRandom(10);
                new PrioritySingleTaskThread().start();
            }
            //sleepRandom(20);
            lowPriorRunReq.setNotify(false);

            for(int j=0; j<Math.random()*20;j++){
                sleepRandom(10);
                new PrioritySingleTaskThread().start();
            }
            //sleepRandom(20);
        }
        mySleep(200);
        System.out.println("Test OK :)))))))))))))))))))))");
        mySleep(200);
        System.exit(0);
    }

    public static void main(String[] args) throws Exception {
        new Main().test();
    }

}
5
  • I'm voting to close this question as off-topic because it's a "review my code" request, not an answerable question. Jun 18, 2015 at 19:28
  • how is review my code not answerable - especially if I ask if that code is thread safe? And I also believe it's quite a repeatable pattern reusable by others. That is in short continuous low priority task pause-able on request by priority task sharing one common resource - I was surprised I didn't find such a code ready made... Jun 18, 2015 at 20:26
  • "Review my code" just isn't a question, even if you try to make it one by putting a question mark after it. Even if you ask a question, but it's about a long bit of real code (not a self-contained example demonstrating an issue), it's still not really a specific question about a specific thing. Here's a simple test: If you can't demonstrate the issue you're asking about with a minimal code example, you probably don't actually have a specific question about a specific issue. Jun 18, 2015 at 20:32
  • Well it is a question how you would define "specific". Believe me actual code has 13K+ lines and I do not want the community to review that ;). What I created was example for pattern I believe may be reusable for others. Recently I start to read a book "head first design patterns" and at the end of each chapter was about same length code. And I think that author would argue that it is a pattern and not a single purpose code. If you strip my code off comment/prints then it is less than 50 lines. The rest of code is for your convenience to play and test it... Jun 18, 2015 at 21:02
  • To be fair, I think this case is close to the line. Other people certainly might disagree and might think it's on the other side of the line. Jun 18, 2015 at 21:09

2 Answers 2

1

I don't know what it means to "start/stop" a "forever thread", but one design pattern you can use for pausing a thread is called a "turnstile". Here's one that works when there is just one controller thread that is allowed to "lock" or "unlock" the turnstile:

import java.util.concurrent.Semaphore;

class Turnstile {
    private final Semaphore semaphore = Semaphore.new(1);

    // Called only from the "controller" thread.
    public void lock() {
        semaphore.acquire();
    }

    // Called only from the "controller" thread.
    public void unlock() {
        semaphore.release();
    }

    // Called from worker threads.
    public void passThrough() {
        semaphore.lock();
        semaphore.release();
    }
}

Initially, the turnstile is in the "unlocked" state, and the passThrough() method promptly returns when ever a worker calls it. If the master thread "locks" the turnstile, then any worker that calls passThrough() will be blocked until the master "unlocks" it again. Then, all of the workers will "pass through", one-by-one.


You could modify this example if you want to have more than one "master", but it's up to you to decide how to resolve the conflict when one master wants the turnstile to be locked, and another wants it to be unlocked.


Then again, instead of modifying Turnstile, you could write a new MultiMasterTurnstile that handles the conflict resolution and uses a Turnstile to actually block the workers.

5
  • Hey it is far off my constrains. But actually it sounds interesting. The main thing is that I have one thread that is most of the time running and I need to tell it to pause (and before de-init shared resource). But it is good idea to centralize all the decision to one class and then this continuous task would only call a method shall_I_Stop() and then shalI_I_Start(). Then High Priority task would call method passThrough(). Yeah it's definitely interesting idea... Thx... I'm pondering if to take a hassle to refactor it... Jun 18, 2015 at 20:41
  • 1
    @VitBernatik, what does it do if it calls shall_I_Start(), and the answer is no? Does it try again? How often? That sounds like polling, and polling often is a bad idea (except if you're writing low-level operating system kernel stuff). Using the Turnstile model, your worker thread calls the passThrough() method at suitable times, and it either will immediately pass through if the turnstile is unlocked, or it will wait there until the turnstile is unlocked. Jun 18, 2015 at 20:46
  • thx it's a good point. If you look to my code you see I'm using wait()-notify() to avoid polling. But you are right the correct would be passThroughHighPriority(), passThroughLowPriority() (instead of shall_I_start). shall_I_stop() would stay. On "yes" it would de-init resource and call passThroughLowPriority() on "no" it would continue to use resource. Probably the code would be very simillar to what I have done but decision would be in one class, hard to guess before I would try to implement... Jun 18, 2015 at 21:09
  • Oh I just noticed there is one big flaw in turnstile. It let threads one by one - but after turnstile they might run concurrent. I would need a double turnstile - letting in only one thread at the time. That is let another thread in when first thread leaves. I will try to do it... Jun 18, 2015 at 21:57
  • Hey so I did rewrite it to DoubleTurnstile and it does look cleaner and also it is more reusable now. Thanks. Still can you please look at my response and tell if it looks to you thread safe? Simple test bench passes so it might be thread safe... Jun 19, 2015 at 9:05
0

As James Large suggested to use Turnstile in his response. I did rewrite the code and it looks nicer at least to me. So I did DoubleTurnstile it looks like this:

public class DoubleTurnstile{
    Semaphore  resourceSemaphore = new Semaphore(1);    // main semaphore protecting resource that has to be accessed sequentially
    volatile boolean  lowPriorRunReq    = false; // low priority task run request
    volatile boolean lowPriorPauseReq   = false; // low priority task pause request (it uses notify)
    private Object notifyWait = new Object();
    Lock       groupLock         = new ReentrantLock();  // group lock (used to acquire lowPriorRunFlag always correctly)
    volatile boolean lowPriorRunFlag = false;  // low priority task run flag

    private void myWaitNotify(){
        synchronized (notifyWait){
            try {
                notifyWait.wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
    private void myNotify(){
        synchronized (notifyWait){
            notifyWait.notify();
        }
    }

    public void highPriorityEnter() {
        groupLock.lock();
        //if (lowPriorRunFlag){ // "if" is not necessary, but correct
            lowPriorPauseReq = true;
        //}
        resourceSemaphore.acquireUninterruptibly();
        groupLock.unlock();
    }

    public void highPriorityLeaves() {
        resourceSemaphore.release();
        groupLock.lock();
        if(lowPriorPauseReq == true){
            lowPriorPauseReq = false;
            myNotify();
        }
        groupLock.unlock();
    }

    public void lowPriorityLoopEnter() {
        while(true){
            if((lowPriorRunReq == true)
                && (lowPriorPauseReq == false)) break;
            myWaitNotify();
        }
        groupLock.lock();
        resourceSemaphore.acquireUninterruptibly();
        lowPriorRunFlag = true;
        groupLock.unlock();
    }

    public boolean lowPriorityLoop_ShallEnd() {
        return (lowPriorRunReq == false)
                || (lowPriorPauseReq == true);
    }
    public void lowPriorityLoop_Leaves() {
        lowPriorRunFlag = false;
        resourceSemaphore.release();
    }

    public void masterLowPriorityRunEn(boolean shallRun) {
        lowPriorRunReq = shallRun;
        myNotify();
    }
}

Then priority task uses DoubleTurnstile as this:

public class PrioritySingleTaskThread extends Thread {
    @Override
    public void run() {
        prn("High Priority Task created");
        dblTurnstile.highPriorityEnter();
        accessResource("high priority task");
        dblTurnstile.highPriorityLeaves();
        prn("High Priority Task closed");
    }
}

and low priority continuous thread uses it like this:

public class LowPriorityContinuousThread extends Thread {
    @Override
    public void run() {
        while(true){
            dblTurnstile.lowPriorityLoopEnter();
            accessResource("low priority init"); // here it is initialization and I want to happen only on request from priority thread
            while(true){
                accessResource("low priority task");
                if(dblTurnstile.lowPriorityLoop_ShallEnd() == true){
                    accessResource("low priority de-init"); // here it is de-initialization and I want to happen only on request from priority thread
                    dblTurnstile.lowPriorityLoop_Leaves();
                    break;
                }

            }
        }
    }
}

Now it is more like pattern and more easily reusable.

Full compilable code with test method is as follows:

import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class Main2 {

    DoubleTurnstile dblTurnstile = new DoubleTurnstile();

    public class DoubleTurnstile{
        Semaphore  resourceSemaphore = new Semaphore(1);    // main semaphore protecting resource that has to be accessed sequentially
        volatile boolean  lowPriorRunReq    = false; // low priority task run request
        volatile boolean lowPriorPauseReq   = false; // low priority task pause request (it uses notify)
        private Object notifyWait = new Object();
        Lock       groupLock         = new ReentrantLock();  // group lock (used to acquire lowPriorRunFlag always correctly)
        volatile boolean lowPriorRunFlag = false;  // low priority task run flag

        private void myWaitNotify(){
            synchronized (notifyWait){
                try {
                    notifyWait.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
        private void myNotify(){
            synchronized (notifyWait){
                notifyWait.notify();
            }
        }
        // call highPriorityLeaves() after used shared resources
        public void highPriorityEnter() {
            groupLock.lock();
            //if (lowPriorRunFlag){ // "if" is not necessary but correct
                lowPriorPauseReq = true;
            //}
            resourceSemaphore.acquireUninterruptibly();
            groupLock.unlock();
        }
        // call exactly once for each previous highPriorityEnter()
        public void highPriorityLeaves() {
            resourceSemaphore.release();
            groupLock.lock();
            if(lowPriorPauseReq == true){
                lowPriorPauseReq = false;
                myNotify();
            }
            groupLock.unlock();
        }

        public void lowPriorityLoopEnter() {
            while(true){
                if((lowPriorRunReq == true)
                    && (lowPriorPauseReq == false)) break;
                myWaitNotify();
            }
            groupLock.lock();
            resourceSemaphore.acquireUninterruptibly();
            lowPriorRunFlag = true;
            groupLock.unlock();
        }

        public boolean lowPriorityLoop_ShallEnd() {
            return (lowPriorRunReq == false)
                    || (lowPriorPauseReq == true);
        }
        public void lowPriorityLoop_Leaves() {
            lowPriorRunFlag = false;
            resourceSemaphore.release();
        }

        public void masterLowPriorityRunEn(boolean shallRun) {
            lowPriorRunReq = shallRun;
            myNotify();
        }
    }

    public class PrioritySingleTaskThread extends Thread {
        int id;
        public PrioritySingleTaskThread(int id){this.id=id;}
        @Override
        public void run() {
            prn("High Priority Task created");
            dblTurnstile.highPriorityEnter();
            accessResource("high priority task",true, id);
            dblTurnstile.highPriorityLeaves();
            prn("High Priority Task closed");
        }
    }

    public class LowPriorityContinuousThread extends Thread {
        public int id = 0;
        @Override
        public void run() {
            while(true){
                dblTurnstile.lowPriorityLoopEnter();
                accessResource("low priority init",false,id++); // here it is initialization and I want to happen only on request from priority thread
                while(true){
                    accessResource("low priority task",false,id++);
                    if(dblTurnstile.lowPriorityLoop_ShallEnd() == true){
                        accessResource("low priority de-init",false,id++); // here it is de-initialization and I want to happen only on request from priority thread
                        dblTurnstile.lowPriorityLoop_Leaves();
                        break;
                    }

                }
            }
        }
    }

    //-------------------------------------------------------------------------
    //-- following functions are meant only for testing
    AtomicInteger clashDetector = new AtomicInteger(0); // only for testing purposes
    int hiPriorityCnt; // only for testing purposes
    int loPriorityCnt; // only for testing purposes
    int lastLowPriorityId=-1;
    int lastHiPriorityId=-1;
    int hiPriorityOutOfOrder=0;

    public void accessResource(String from,boolean hiPriority, int id) {
        prn("Resource used from " + from+" id: "+id);
        if(hiPriority){
            if( (id - lastHiPriorityId) < 1) {
                // note if id - lastHiPriorityId=+2 (one sample over-jumped) it will be detected
                // when returned to the over-jumped sample,
                // or at the end of the test one sample missing will be detected
                // so no injustice will escape it's punishment ;)
                // On the other hand if we want strictly ==1 then one error will be reported 3 times -
                // 1st when ID: 1->3
                // 2nd when ID: 3->2 // correct error
                // 3rd when ID: 2->4
                System.out.println("High priority jumped over each other - it's not nice but it can happen");
                hiPriorityOutOfOrder++;
            }
            lastHiPriorityId = id;
            hiPriorityCnt++;
        }
        else{
            if( (id - lastLowPriorityId) < 1) {
                System.out.println("LowPriorityLoop request swapped - you are a bad programmer :((((((");
                System.exit(-1);
            }
            lastLowPriorityId = id;
            loPriorityCnt++;
        }
        if (clashDetector.addAndGet(1) > 1) {
            System.out.println("Clash detected - you are a bad programmer :((((((");
            System.exit(-1);
        }
        sleepRandom(5);
        clashDetector.getAndAdd(-1);
    }

    public void sleepRandom(long maxMiliSec) {
        mySleep((long) (Math.random() * maxMiliSec));
    }

    public void mySleep(long miliSec) {
        try {
            Thread.sleep(miliSec);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    void prn(String s) {
        System.out.println(s);
    }

    public void test() {
        int idHiPriority = 0;
        LowPriorityContinuousThread lowPriorThrd = new LowPriorityContinuousThread();
        lowPriorThrd.start();
        for (long i = 0; i < (long) 1e3; i++) {
            dblTurnstile.masterLowPriorityRunEn(true);
            for (int j = 0; j < Math.random() * 100; j++) {
                sleepRandom(10);
                new PrioritySingleTaskThread(idHiPriority++).start();
            }
            //sleepRandom(20);
            dblTurnstile.masterLowPriorityRunEn(false);

            for (int j = 0; j < Math.random() * 20; j++) {
                sleepRandom(10);
                new PrioritySingleTaskThread(idHiPriority++).start();
            }
            //sleepRandom(20);
        }
        mySleep(500);
        boolean testOk = true;
        if(hiPriorityCnt != idHiPriority){
            System.out.println(String.format("Error hiPriorityCnt(%d) != idHiPriority(%d)",
                    hiPriorityCnt, idHiPriority));
            testOk = false;
        }
        if(loPriorityCnt != lowPriorThrd.id){
            System.out.println(String.format("Error loPriorityCnt(%d) != lowPriorThrd.id(%d)",
                    loPriorityCnt, lowPriorThrd.id));
            testOk = false;
        }
        System.out.println("High priority tasks performed: "+hiPriorityCnt);
        System.out.println("High priority out of order: "+hiPriorityOutOfOrder);
        System.out.println("Low priority tasks performed: "+loPriorityCnt);
        if(testOk){
            System.out.println("Test2 OK :)))))))))))))))))))))");
        }else{
            System.out.println("Test Failed :(((((((((((((((((((((");
        }
        mySleep(200);
        System.exit(0);
    }

    public static void main(String[] args) throws Exception {
        new Main2().test();
    }
}

The logic of the locks did not change too much. So I still would appreciate if you leave a comment if you think that my DoubleTurnstile class is thread safe under all conditions. Tests suggests so.

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.