Algorithm - execute task only for unique entries in queue, common entries should wait

Question

We are creating a rest application. And we have an edge condition where parallel actions are not supported on same object. For example :

Not supported in parallel
    Request 1 for action XYZ for object  A
    Request 2 for action XYZ for object  A
    Request 3 for action ABC for object  A

Supported in parallel
    Request 1 for action XYZ for object  A
    Request 2 for action XYZ for object  B
    Request 3 for action ABC for object  C

Now, the object count is not fixed. we can have n number of such objects.

I want that if a request for object A is under progress then other request for object A should wait for existing task on object A to get over.

But I am not able to figure out the algorithm for this purpose.

I could plan for below design but not able to figure out on how to use the locking since all objects can be different.

• A queue which stores the entry for object A when request comes.
• Entry gets deleted if response is sent
• If an entry is already present, then wait for existing request to get over.
• If entry not present, then execute immediately.

Now task on object A should not impact the task on object B. So they must accept unique locks.

And also, request cannot go standalone and be queued. Somehow I have to make the current thread sleep so that I can send response to user.

Can anyone guide here?

Answer 1

UPDATED based on comments from my original response

The ideal model for something like that would be using an actor system such as Akka.

But your comment states that this will happen in the context on a REST application where threads will be blocked already by request processing.

In this case, the idea would be using a per-object-guard such as:

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CountDownLatch;

public class ObjectGuard<K> {
    private final ConcurrentMap<K, CountDownLatch> activeTasks = new ConcurrentHashMap<>();

    public Guard guardFor(final K key) throws InterruptedException {
        if (key == null) {
            throw new NullPointerException("key cannot be null");
        }

        final CountDownLatch latch = new CountDownLatch(1);

        while (true) {
            final CountDownLatch currentOwner = activeTasks.putIfAbsent(key, latch);
            if (currentOwner == null) {
                break;
            } else {
                currentOwner.await();
            }
        }

        return () -> {
            activeTasks.remove(key);
            latch.countDown();
        };
    }

    public interface Guard extends AutoCloseable {
        @Override
        void close();
    }
}

You would use it as follows:

class RequestProcessor {
    private final ObjectGuard<String> perObjectGuard = new ObjectGuard<>();

    public String process(String objectId, String op) throws InterruptedException {
        // Only one thread per object id can be present at any given time
        try (ObjectGuard.Guard ignore = perObjectGuard.guardFor(objectId)) {
            String result = ... // compute response
        }
    }
}

If two concurrent calls to process are received for the same object id, only one will be processed, the others wait their turn to process a request on that object.

Answer 2

Signalling like juancn does in his answer is not my strong suit, so I made an even cruder solution using one Semaphore for signalling combined with a request-counter.

There is one lock involved (subjectsLock) which synchronizes everything at one point in time. The lock is required to ensure there are no memory leaks: since there can be any number of subjects (a.k.a. object identifiers in your question), cleanup is essential. And cleanup requires knowing when something can be removed and that is difficult to determine without a lock that brings everything to one known state at a certain point in time.

The test in the main-method in the code shown below is a bit hard to read, but it serves as a starting point for a demonstration of how the code works internally. The main logic is in the methods executeRequest, addSubject and removeSubject. If those three methods do not make sense, another solution should be used.

Stress-testing will have to determine if this solution is fast enough: it depends on the number of requests (per second) and the amount of time it takes to complete an action. If there are many requests and the action is short/fast, the (synchronization) overhead from the lock could be to high.

// package so;

import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantLock;
import java.util.stream.IntStream;

public class RequestQueue {

    public static void main(String[] args) {

        // Randomized test for "executeRequest" method below.

        final int threadCount = 4;
        ExecutorService threadPool = Executors.newFixedThreadPool(threadCount);
        try {
            final int requestCount = 100;
            final RequestQueue rq = new RequestQueue();
            final Random random = new Random();
            IntStream.range(0, requestCount).forEach(i -> threadPool.execute(new Runnable() {
                @Override
                public void run() {
                    try {
                        String subject = "" + (char) (((int)'A') + random.nextInt(threadCount));
                        rq.executeRequest(subject, new SleepAction(i, subject, 50 + random.nextInt(5)));
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }));
            sleep(100); // give threads a chance to start executing.
            while (true) {
                sleep(200);
                List<String> subjects = rq.getSubjects();
                System.out.println("Subjects: " + subjects);
                if (subjects.isEmpty()) {
                    break;
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            threadPool.shutdown();
        }
    }

    private Map<String, QueueLock> subjects = new LinkedHashMap<>();

    // a fair ReentrantLock is a little bit slower but ensures everybody gets their turn in orderly fashion.
    private final ReentrantLock subjectsLock = new ReentrantLock(true);

    private class QueueLock {

        // a fair Semaphore ensures all requests are executed in the order they arrived.
        final Semaphore turn = new Semaphore(1, true);
        final AtomicInteger requests = new AtomicInteger(1);

        public String toString() { return "request: " + requests.get(); }
    }

    /**
     * Allow all requests for different subjects to execute in parallel,
     * execute actions for the same subject one after another.
     * Calling thread runs the action (possibly after waiting a bit when an action for a subject is already in progress).
     */
    public String executeRequest(String subject, Runnable action) throws InterruptedException {

        QueueLock qlock = addSubject(subject);
        try {
            int requestsForSubject = qlock.requests.get(); 
            if (requestsForSubject > 1) {
                System.out.println(action.toString() + " waiting for turn " + requestsForSubject);
            }
            qlock.turn.acquire();
            if (requestsForSubject > 1) {
                System.out.println(action.toString() + " taking turn " + qlock.requests.get());
            }
            action.run();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            removeSubject(subject);
        }
        return timeSinceStart() + " " + subject;
    }

    private QueueLock addSubject(String s) {

        QueueLock qlock = null;
        subjectsLock.lock();
        try {
            qlock = subjects.get(s);
            if (qlock == null) {
                qlock = new QueueLock();
                subjects.put(s, qlock);
            } else {
                qlock.requests.incrementAndGet();
            }
        } finally {
            subjectsLock.unlock();
        }
        return qlock;
    }

    private boolean removeSubject(String s) {

        boolean removed = false;
        subjectsLock.lock();
        try {
            QueueLock qlock = subjects.get(s);
            if (qlock.requests.decrementAndGet() == 0) {
                subjects.remove(s);
                removed = true;
            } else {
                qlock.turn.release();
            }
        } finally {
            subjectsLock.unlock();
        }
        return removed;
    }

    public List<String> getSubjects() {

        List<String> subjectsBeingProcessed = new ArrayList<>();
        subjectsLock.lock();
        try {
            // maintains insertion order, see https://stackoverflow.com/a/18929873/3080094
            subjectsBeingProcessed.addAll(subjects.keySet()); 
        } finally {
            subjectsLock.unlock();
        }
        return subjectsBeingProcessed;
    }

    public static class SleepAction implements Runnable {

        final int requestNumber;
        final long sleepTime;
        final String subject;

        public SleepAction(int requestNumber, String subject, long sleepTime) {
            this.requestNumber = requestNumber;
            this.sleepTime = sleepTime;
            this.subject = subject;
        }

        @Override
        public void run() {

            System.out.println(toString() + " sleeping for " + sleepTime);
            sleep(sleepTime);
            System.out.println(toString() + " done");
        }

        public String toString() {return timeSinceStart() + " " + subject + " [" + Thread.currentThread().getName() + "] " + String.format("%03d",requestNumber); }

    }

    public static final long START_TIME = System.currentTimeMillis();
    public static String timeSinceStart() {
        return String.format("%05d", (System.currentTimeMillis() - START_TIME));
    }

    public static void sleep(long milliseconds) {

        try {
            Thread.sleep(milliseconds);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}

Answer 3

An object which executes requests serially is known as Actor. The most widely known java actor library is named Akka. The most simple (one page) actor implementation is my SimpleActor.java.