How to keep the order of executing tasks? - Help needed.

Joined
Feb 21, 2023
Messages
1
Reaction score
0
A problem that I try to understand and solve states the following:

There is a set of tasks, each with some index targetId, and this set has to be executed concurrently in an efficient way. The Executor cannot execute tasks with the same targetId concurrently, but it can do so for many tasks with different targetId. A limit on the number of concurrent threads is also given (maxThreads; if 0, it's assumed there's no limit).

The program that attempts to do so is (https://codeshare.io/PdRZ1o). However, sometimes it fails to pass some tests. The issue is the program must execute tasks with the same targetId in the exact order they appear in the queue (which can be filled up during the execution). How to modify the code?
 
Joined
Mar 5, 2023
Messages
36
Reaction score
12
To modify the code to ensure tasks with the same targetId are executed in the exact order they appear in the queue, you can use a HashMap to store the list of tasks with the same targetId. When a new task is added to the queue, check if there are any other tasks with the same targetId already in the HashMap. If yes, then add the new task to the list of tasks with the same targetId. If not, then create a new entry in the HashMap for the new targetId and add the task to the list.

When a task is dequeued, check if there are any other tasks with the same targetId in the HashMap. If yes, then dequeue the next task with the same targetId and execute it. If not, then execute the dequeued task as usual.

Here is an updated version of the code with the above modifications:

Java:
import java.util.*;
import java.util.concurrent.*;

public class ExecutorImpl implements Executor {

    private final Queue<Runnable> taskQueue = new ConcurrentLinkedQueue<>();
    private final Map<Integer, LinkedList<Runnable>> taskMap = new HashMap<>();
    private final int maxThreads;

    public ExecutorImpl(int maxThreads) {
        this.maxThreads = maxThreads;
    }

    @Override
    public void execute(Runnable command) {
        synchronized (taskQueue) {
            taskQueue.add(command);
        }
        scheduleTasks();
    }

    private void scheduleTasks() {
        while (true) {
            if (maxThreads > 0 && Thread.activeCount() >= maxThreads) {
                return;
            }
            Runnable task = null;
            synchronized (taskQueue) {
                task = taskQueue.poll();
            }
            if (task == null) {
                return;
            }
            Integer targetId = ((Task) task).getTargetId();
            LinkedList<Runnable> targetTasks;
            synchronized (taskMap) {
                targetTasks = taskMap.get(targetId);
                if (targetTasks == null) {
                    targetTasks = new LinkedList<>();
                    taskMap.put(targetId, targetTasks);
                }
            }
            synchronized (targetTasks) {
                targetTasks.add(task);
            }
            if (Thread.activeCount() == 1) {
                new Thread(this::executeTasks).start();
            }
        }
    }

    private void executeTasks() {
        while (true) {
            Map.Entry<Integer, LinkedList<Runnable>> nextTaskEntry = null;
            synchronized (taskMap) {
                for (Map.Entry<Integer, LinkedList<Runnable>> entry : taskMap.entrySet()) {
                    LinkedList<Runnable> tasks = entry.getValue();
                    if (!tasks.isEmpty()) {
                        nextTaskEntry = entry;
                        break;
                    }
                }
            }
            if (nextTaskEntry == null) {
                return;
            }
            LinkedList<Runnable> targetTasks = nextTaskEntry.getValue();
            Runnable nextTask;
            synchronized (targetTasks) {
                nextTask = targetTasks.poll();
            }
            if (nextTask == null) {
                continue;
            }
            nextTask.run();
        }
    }
}

class Task implements Runnable {
    private final int targetId;
    private final int taskId;

    public Task(int targetId, int taskId) {
        this.targetId = targetId;
        this.taskId = taskId;
    }

    public int getTargetId() {
        return targetId;
    }

    public int getTaskId() {
        return taskId;
    }

    @Override
    public void run() {
        System.out.println("Task " + taskId + " with targetId " + targetId + " started");
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            return;
        }
        System.out
 

Ask a Question

Want to reply to this thread or ask your own question?

You'll need to choose a username for the site, which only take a couple of moments. After that, you can post your question and our members will help you out.

Ask a Question

Members online

No members online now.

Forum statistics

Threads
473,818
Messages
2,569,727
Members
45,661
Latest member
NadineBour

Latest Threads

Top