N threads enter, one and only one should do extra processing that others wait for

Question

My task is, I think, pretty interesting, if a bit unusual. Any number of threads can call function f(). They add some data to a buffer and begin waiting (the function is blocking). Eventually some extra processing has to be done on the whole buffer, and all the previous calls to f() wait for this processing. Either the buffer fills up, and the thread that filled it up can call process() - this is easy to implement. Or - this is what's giving me a headache - some time passes and process() should be called on timeout. Either way, all the calls to f() can only return after process() completed. The problem is handling the timeout such that one and only one thread calls process() (doesn't matter which one), and the others wait for it to finish.

I have spent at least half an hour thinking it through and didn't find any good implementation. By good I mean not extremely convoluted, and if possible - it should be obviously correct in terms of inter-thread synchronization. How would you do it?

I'm coding in C++(20/23), a lower-level solution (e. g. using pthreads or OS syscalls) is also very welcome.

Answer 1

Here's an example that I think works, more-or-less. I think there's a race-condition on writing at the same time the timeout occurs. You didn't ask about it, but I find that these issues tend to come to light once you start building a working solution which is why I mention it. You might need another barrier of some sort on writes to the buffer. In the last update, I added one possible solution for that and made the writer threads more chaotic.

Because these answers may be useful to others in the future, I'll give an overview of the concurrency primitives used here. Feel free to skip this section if you are familiar with the terms. The terminology is pretty standard (AFAIK) and not specific to Java. I link to the old 1.8 JavaDocs because these concepts don't really change much and they seem to adequately describe them.

• Semaphore In this example, a semaphore is used to allow only one thread to enter a 'critical section' at a time. The limit can be whatever you want but I'm using them to force 'single-file' access. These are bottlenecks and could be points of contention. I think of the semaphore as being like a 'talking stick'.
• CountDownLatch A little less common, perhaps, than a semaphore, a count down latch is something that I use to block threads until some sort of state has been achieved. The best analogy I can think of at the moment is like the gates at a horse race. The jockeys bring their horses to the gates and when the race begins all the gates open at the same time. That's like what happens when the latch hits zero. All the threads waiting 'take-off', or at least, that's what you should assume.
• AtomicBoolean I prefer these types to their primitive equivalents marked as volatile mainly because I think the semantics are easier to reason about and that's always an advantage when worrying about multithreaded code. Yes, 'worrying'. You don't write multithreaded code as much as you worry about it. I might be able to replace these with volatile booleans but, why? It probably won't make it any better and it could be a lot worse. I'm not sure what this would translate to in C++, consult the documentation.

There is one CountDownLatch in the example below named hold. All the writer threads will wait on this if the buffer is not filled during their writing to it. When the countDown method is called on it, all the threads waiting on it will be released. While they may not all 'wake' at the exact same moment, it's crucial to assume that they do when designing with this kind of primitive.

There are two Semaphores in the design both are used force only one thread to be executing critical sections of code at a time. The writeLock prevents multiple threads from writing to the buffer at the same time. It is also used to make sure that no thread writes anything to the buffer after processing has started. That is, it also protects the done flag. Without that protection, a thread could enter write around the same time as the timer released the threads waiting on hold and nothing would prevent the process and write methods to be executing at the same time.

The processorLock semaphore is used in a similar way as the writeLock except that it simply prevents multiple threads from running the processor method at the same time, as per your requirements. When the hold is released and the 'horses' (threads) leap from their gates, only one of them can acquire this lock at a time. Whichever one does will do the processing (waiting for any thread in the write method) and mark the done flag. All the other threads will, one-at-a-time, check the done flag and exit.

I do not claim this to be optimal and there are some flaws that I can think of such as the possbility that new threads keep writing to the buffer (until it is full) and preventing processing after the timeout has expired. Depending on your requirements, that might be an issue, or maybe it isn't. You could probably use a single semaphore for both process locking and write locking. That has a risk of contention, in general, but in the current configuration of the example below, I don't think it makes much difference with regard to contention either way. I've left them as separate because I think it might be informative.

Hopefully this will get you started thinking about these concerns and how you might deal with them. All multi-threaded code should be expected to have bugs, so I invite all questions and comments.

This code does run in VSCode (on my machine ;-) so please play around with timeouts, threads, etc. You just need to put it in a folder called 'concurrency' and name the file 'Demo.java'.

package concurrency;

import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.Random;

import java.util.Timer;
import java.util.TimerTask;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicBoolean;


class Demo {
    private static final String[] badBoys = {"bad", "boys", "bad", "boys", "whatcha", "gonna", "do", "whatcha", "gonna", "do", "when", "they", "come", "for", "you"};
    private static final Random random = new Random();
    private static final int MAX_SLEEP = 500;

    public static void main(String... args) {
        final int buffer_limit = 1500;
        final long timeout_ms = 5000;
        final long threads = 50;
        
        BufferContext context = new BufferContext(buffer_limit, timeout_ms);

        for (int i = 0; i < threads; i++) {
            Thread thread = new Thread(new BufferThread(context, badBoys));
            thread.start();
        }

        context.startTimer();
    }

    public static void print(String s) {
        System.out.println(Thread.currentThread() + ": " + s);
    }

    public static void random_sleep() throws InterruptedException {
        Thread.sleep(random.nextInt(MAX_SLEEP));
    }
}

class BufferContext {
    private final List<String> buffer = new CopyOnWriteArrayList<>();
    private final Semaphore writeLock = new Semaphore(1);
    private final CountDownLatch hold = new CountDownLatch(1);
    private final Semaphore processorLock = new Semaphore(1);
    private final AtomicBoolean done = new AtomicBoolean(false);
    private final Timer timer = new Timer();
    private final int limit;
    private final long timeout;

    BufferContext(int limit, long timeout) {
        this.limit = limit;
        this.timeout = timeout;
    }

    boolean add(String s) throws InterruptedException {
        writeLock.acquire();
        try {
            if (done() || full()) {
                return false;
            } else {
                buffer.add(s);
                return true;
            }
        } finally {
            writeLock.release();
        }
    }

    boolean done() {
        return done.get();
    }

    boolean full() {
        return buffer.size() >= limit; 
    }

    void await() throws InterruptedException {
        Demo.print("waiting");
        hold.await();
    }

    void release() {
        hold.countDown();
    }

    void process() throws InterruptedException {
        processorLock.acquire();

        try {
            if (!done.get()) {
                writeLock.acquire();
                done.set(true);
                writeLock.release();
                timer.cancel();

                Demo.print("processing buffer");
                for (String s : buffer) {
                    System.out.print(s);
                    System.out.print(' ');
                }
                System.out.println();
            }
        } finally {
            processorLock.release();
        }
        
        release();
        
        Demo.print("finished");
    }

    void startTimer() {
        TimerTask task = new TimerTask(){
            @Override
            public void run() {
                Demo.print("timer done");
                release();
            }
        };
        
        timer.schedule(task, timeout);
    }
}

class BufferThread implements Runnable {
    private final String[] data;
    private final BufferContext context;

    public BufferThread(BufferContext context, String... data) {
        this.data = data;
        this.context = context;
    }

    @Override
    public void run() {
        try {
            for (String s : data) {
                Demo.random_sleep();
                if (!context.add(s)) break;
            }
            
            if (!context.full()) {
                context.await();
            } else {
                Demo.print("buffer full");
            }

            context.process();
        } catch (InterruptedException e) {
            // I think this is irrelevant, and I can't be bothered with it.
        }
    }    
}

Answer 2

Perhaps you can solve this by adding an extra thread (or by scheduling a callback in a thread pool, etc.) responsible for the timeout.

These seem to be cooperating threads, right? So the first adding data starts the timeout. The threads that fills the buffers does the "final processing". Or the thread handling the timeout does it, whichever comes first. Be sure to have a critical section (or mutex) to access the buffer and control structures.

Answer 3

If the goal is simplification of the algorithm then first create the logic for a "normal" thread to:

• Add it's message to the buffer
• Check the timeout value
• If enough time has passed AND the buffer is not empty call process(), reset the time, and release the waiting threads
• Otherwise, wait

Then, treat the timer-spawned cleanup thread as a normal thread with an empty message. It will contribute nothing to the buffer, call process(), reset the timer, and release any waiting threads.

The trick here is to simplify/unify the logic for the "normal" threads and the "timer" thread.

NOTE #1: It will be necessary to handle messages that fill or overflow the buffer. However, the design idea of one code path for normal and timer threads still applies.

NOTE #2: Regarding efficiency I'd try to find a non-blocking approach for this.