4

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.

11
  • @user253751: an interesting idea for sure, but once_flag cannot be reset. Jan 17, 2023 at 17:42
  • What's the relationship (temporal, call/callee, etc.) between calling f() and adding data to the buffer?
    – Pablo H
    Jan 17, 2023 at 17:59
  • 2
    Is there any particular reason why the thread which calls process() has to be one of the threads which called f()? It sounds to me as if process() makes more sense on an entirely separate thread whose responsibility is for deciding if/when to call process() and act as a consumer/controller for the buffer. Jan 17, 2023 at 18:32
  • @BenCottrell: you're right, and there is no reason in terms of functionality, only for efficiency. And the concept itself seems interesting enough to explore. Jan 17, 2023 at 19:23
  • When a thread calls the buffer as it's filling up what does the thread do with the data that it can't send to the buffer since it filled it up? Jan 17, 2023 at 19:43

3 Answers 3

2

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.
        }
    }    
}
5
  • Thank you! I get the idea, but the intricacies of the inter-thread synchronization are hard for me to follow in some spots. I'm sure this code is correct, but that's not obvious. I think Pablo's idea of designating one thread responsibe for handing the timeout is the key to reducing complexity. Jan 19, 2023 at 23:05
  • 1
    P. S. Kudos for creating an actual runnable example, complete with main()! Jan 19, 2023 at 23:06
  • 1
    @VioletGiraffe There is actually a thread dedicated to that in the above solution. It's implicitly created by the Timer class. I think it's a good idea which is why I borrowed it but by itself, I don't see how it solves the larger problem. That is you will end up with race conditions such as a thread filling up the buffer at the same time the timeout occurs. The solution here errs on the side of correctness. It can surely be improved and if you have any questions, I can walk you through it. Would a textual description or more comments help?
    – JimmyJames
    Jan 20, 2023 at 17:17
  • I think I understand it for the most parts, but some comments would be nice to see in the beginning of process() where locks are acquired and conditions are checked, e. g. is the order important for correctness. As far as I understand, writerLock should make sure there is no race condition in the case of one thread adding to the buffer while another is processing it. Jan 20, 2023 at 17:29
  • 1
    @VioletGiraffe writerLock is serving two purposes in the current update. It's making it so that no two threads can try to write at the buffer at the same time which you may or may not need. The main reason I added it was to avoid the race condition I mentioned in my last comment. That is, if a thread is writing when the timer fires, you could end up a with one thread writing to the buffer while the other is processing (or after it is done.) I'll add some commentary to the answer to explain things a bit and some comments.
    – JimmyJames
    Jan 20, 2023 at 18:57
1

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.

1
  • This can be solved by an extra thread for sure, and quite easily in terms of how the code would look like, but it's also less efficient. Your second idea is very interesting - designate one thread responsible for the timeout, I have not considered that! Jan 17, 2023 at 19:22
1

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.

10
  • I think you forgot the requirement that if a thread fills the buffer, it should execute the processing immediately. Also, w/ regard to note 2, that this is blocking is an explicit requirement according to the OP.
    – JimmyJames
    Jan 23, 2023 at 15:31
  • Also, the timeout could expire but nothing will happen until another writing thread comes along.
    – JimmyJames
    Jan 23, 2023 at 15:34
  • There's also the race condition when two threads are writing when the timeout occurs and other similar scenarios.
    – JimmyJames
    Jan 23, 2023 at 15:36
  • @JimmyJames Bullet #3 and note #1 address your first and second comments. Race conditions (and deadlock) may occur in concurrent designs. Your third comment is correct, I did not address race conditions (or deadlock) in this answer. Jan 23, 2023 at 20:48
  • 1
    All good. You could say I'm aware of the challenges.... Jan 23, 2023 at 22:01

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.