Semaphore skipping additional threads

Question

In this example I've a refresh method that might be called by multiple threads. Requirements are:

1. refresh() method must be processed by only one thread at a time (blocking)
2. If already a thread is waiting, skip any new thread tries to acquire the lock to avoid processing multiple times

The refresh() method not returning anything nor changing state for the calling thread (fire and forget). So it is not necessary that it's called for every thread, instead, if multiple thread ask for a refresh it should only be executed once.

I'm trying to use a Semaphore with 1 permit for this task. If the semaphore has any queued threads I'm just returning, otherwise the thread will acquire the lock.

class Scratch {

    private final Semaphore refreshLock = new Semaphore(1);

    public void refresh() {
        System.out.println("call refresh()");
        try {
            synchronized (refreshLock) {
                if (refreshLock.hasQueuedThreads()) {
                    System.out.println("skip this thread ... " + Thread.currentThread().getId());
                    return; // don't put one more to the queue
                }
            }

            // wait till release of lock
            System.out.println("wait for lock release, queue: " + refreshLock.getQueueLength());
            refreshLock.acquire();

            try {
                System.out.println("process ... " + Thread.currentThread().getId());
                Thread.sleep(5000);
                System.out.println("### DONE");
            } finally {
                refreshLock.release();
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    static class Task implements Runnable {
        private int taskId;
        private Scratch scratch;

        Task(int id, Scratch scratch) {
            this.taskId = id;
            this.scratch = scratch;
        }

        @Override
        public void run() {
            System.out.println("Task ID : " + this.taskId + " performed by " + Thread.currentThread().getName());
            try {
                Thread.sleep(300 + taskId * 10);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            scratch.refresh();
        }
    }

    public static void main(String[] args) {
        Scratch scratch = new Scratch();
        ExecutorService service = Executors.newFixedThreadPool(10);
        IntStream.range(0, 12).forEach(i -> service.submit(new Task(i, scratch)));
        service.shutdown();
    }
}

Answer 1

refresh() method must be processed by only one thread at a time ...

private final Semaphore refreshLock = new Semaphore(1);

A mutex is arguably a better synchronization object choice when you only need one thread at a time. So I would advise you to use a ReentrantLock instead, for instance.

synchronized (refreshLock) {
    if (refreshLock.hasQueuedThreads()) {
        System.out.println("skip this thread ... " + Thread.currentThread().getId());
        return; // don't put one more to the queue
    }
}

So in fact you are using not one but two synchronization objects, a semaphore and a monitor. The reason why you are doing this is because of your second requirement. However you can implement both of your requirements using only one synchronization object which would make your code simpler (hopefully).

Thread.sleep(5000);

No! I know you are doing this to emulate a long running operation. However long running operations should never execute within concurrent control, specially if it's running one at a time.

What this means effectively is that all threads calling refresh will be blocked by at least a whopping 5 seconds.

So obviously you have to implement your refresh method as a process that runs on the background. Meaning that you will be returning a Future that may complete at any given moment. For a thread that will do the process this moment will be when the process has finished, for all others this will be almost immediately.

This is the resulting implementation:

private final ReentrantLock refreshLock = new ReentrantLock();
private boolean hasRefreshed = false;
ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors.newFixedThreadPool(2);

public Future refresh() {
    System.out.println("call refresh()");
    try {
        refreshLock.lock();
        if(hasRefreshed) {
            System.out.println("already refreshed");
            return executor.submit(() -> {});
        }

        hasRefreshed = true;
        return executor.submit(() -> {
            System.out.println("process ... " + Thread.currentThread().getId());
            try {
                Thread.sleep(5000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("### DONE");
        });

    }finally {
        refreshLock.unlock();
    }
}

Ideally the synchronization logic would be implemented in a generic class that would be able to handle any use case.

You will also have to change your main a bit, so your program doesn't exit immediately.

IntStream.range(0, 12)
    .mapToObj(i -> service.submit(new Task(i, scratch)))
    .forEach(f -> {
        try {
            f.get();
        } catch (InterruptedException e1) {
            // TODO Auto-generated catch block
            e1.printStackTrace();
        } catch (ExecutionException e1) {
            // TODO Auto-generated catch block
            e1.printStackTrace();
        }
    });

Answer 2

The same effect can be achieved with a Queue and a DiscardPolicy.

    int poolSize=1;
    int maxPoolSize=1;
    int queueSize=1;
    long aliveTive=1000;  
    ArrayBlockingQueue<Runnable> queue = 
            new ArrayBlockingQueue<Runnable>(queueSize);
    ThreadPoolExecutor executor = 
            new ThreadPoolExecutor(poolSize, maxPoolSize, aliveTive,
                    TimeUnit.MILLISECONDS,queue,new ThreadPoolExecutor.DiscardPolicy());

Essentially, only one task can remain waiting in the queue. A second waiting task cannot be added - the new one will be discarded. (Dropping the old one and accepting the new one is also possible with the DiscardOldestPolicy.)

Example from here: https://stackoverflow.com/questions/30299784/using-threadpoolexecutor-and-discardpolicy

More details about RejectedExecutionHandler and different built-in policies: https://docs.oracle.com/javase/7/docs/api/java/util/concurrent/ThreadPoolExecutor.html