Java, Thread and Priorities

Question

I'm working on a problem, that was supposed to be VERY simple to solve, however I am not getting it done so easily.

The problem is quite simple: I have a Java Program running on Linux/x86 that can perform two basic functionalities F1 and F2. I would like to set F1 to have a higher priority, even though it is a MUST that F2 executes from times to times, i.e, the fact of having F1 requests on queue cannot put F2 requests waiting forever.

My first though was just having separate queues with a thread pool for each functionality, I set the F1 pool to have 8 threads while the F2 pool got only 2 threads.

On my expectaion linux would give fairly time share for each thread, so F1 would have 8 quantums while F2 would get just 2. If there was no F1 requests, F2 pool could get every quantum to itself, the same should be true for F1 just in case F2 has no requests.

However, the program is not behavingthat way, if I get a burst of F2 requests and just couple of F1 requets, the latter is taking a long time to get its turn.

Doest that make sense talking about Oracle HotSpot/linux scheduling? Or it should not be happening, what would point to an implementation error from my part?

PS: I've read about linux scheduling, and it seems that SCHED_OTHER (TS) gives time share for each task, however every time a task ready is not executed it gets a bigger quantum, and if that is happening to F2 pool, that might explain the above mentioned behavior.

Thanks and Regards.

Below there is a sample source code.

package test;

import java.util.Properties;
import java.util.concurrent.ArrayBlockingQueue;

/**
 * Created by giscardff on 08/07/18.
 */
public class TestThread {

    // Test Program
    public static void main(String args[]) throws Exception {

        // queues containing jobs to be done
        ArrayBlockingQueue<MyDTO> queueA = new ArrayBlockingQueue<>(100);
        ArrayBlockingQueue<MyDTO> queueB = new ArrayBlockingQueue<>(100);

        // create pool for functionality A
        for(int i = 1; i <= 8; i++){
            MyThread thread = new MyThread("ThreadA" + i, queueA);
            thread.start();
        }

        // create pool for functionality B
        for(int i = 1; i <= 2; i++){
            MyThread thread = new MyThread("ThreadB" + i, queueB);
            thread.start();
        }

        // create producer for A
        // it will take 100ms  between requests
        Producer producerA = new Producer(queueA, 0);
        producerA.start();

        // create producer for B
        // it will take 0ms between requests
        Producer producerB = new Producer(queueB, 0);
        producerB.start();

    }

}

/**
 * Just put a request into a queue
 */
class Producer extends Thread {

    private ArrayBlockingQueue<MyDTO> queue;
    private long sleep;

    public Producer(ArrayBlockingQueue<MyDTO> queue, long sleep){
        this.queue = queue;
        this.sleep = sleep;
    }

    @Override
    public void run() {
        try {
            while (true) {
                if(sleep > 0)Thread.sleep(sleep);
                queue.put(new MyDTO());
            }
        }catch(Exception ex){}
    }
}

/**
 * Retrieve a request from a queue, calculate how long request took to
 * be received for each 1M requests
 */
class MyThread extends Thread {

    private ArrayBlockingQueue<MyDTO> queue;
    private long delay = 0;
    private int count = 0;

    public MyThread(String name, ArrayBlockingQueue<MyDTO> queue){
        super(name);
        this.queue = queue;
    }

    @Override
    public void run() {
        try {
            while (true) {
                MyDTO input = queue.take();
                delay += System.currentTimeMillis() - Long.parseLong(input.getTime());
                if(++count % 1000 == 0){
                    System.out.printf("%s: %d\n", getName(), delay / 10);
                    count = 0;
                }
            }
        }catch(Exception ex){ex.printStackTrace();}
    }
}

/**
 * Just a DTO representing a request
 * NOTE: The time was set as String to force CPU to do something more than just math operations
 */
class MyDTO {
    private String time;
    public MyDTO(){
        this.time = "" + System.currentTimeMillis();
    }

    public String getTime() {
        return time;
    }
}

Answer 1

It looks like you've got a few issues. I'll try to summarize them and provide starting point for a path forward:

Thread Contention

Using the BlockingQueue comes with a cost - every write operation (put & take) is lock contended between the producers or consumers. Your "A pool" has 9 threads contending over the write lock for queueA (1 producer, 8 consumers), while your "B pool" has 3 threads contending over the lock for queueB (1 producer, 2 consumers).

This related answer provides a bit more detail about contention. The simplest ways around this are to "use less threads" or use "lock-free" mechanisms to eliminate the contention.

Thread Scheduling

As mentioned in the comments, you're at the mercy of how the JVM is scheduling your threads.

If java thread scheduling used perfectly fair time shares on the CPU, you'd probably see consumption counts on each thread in the same pool extremely close to each other. You've probably noticed they're not - my runs of your (slightly modified) code occasionally give me a count spread of 300K or more across the threads.

You can often get this better when there are enough CPU cores for each CPU-bound thread (you've got 12 in your sample code), but it's far from ideal in many cases, especially in the face of thread contention.

What can you do?

1. Build your own fairness logic - don't rely on the JVM thread scheduler to be fair, because it won't be.
   • A simple idea for your case would be to keep both queues, but use a single pool to process both - use either round-robin or Math.random() (ie: if (rand < 0.8) { queueA.poll();}) to determine which queue to poll from. Note - Use poll so you can easily handle the case when a queue is empty without blocking.
2. Experiment with number of CPU-bound threads running on your hardware. With my suggestion for (1) above, you could even have just one worker thread fairly processing both queues. Remember, too many threads contending over the same resources will slow down your processing.

Isn't threading fun? :)