Prioritize task update based on their status

[mcornaton]

No description provided.

[mcornaton]

If we ever need it, this could be easily updated to add deadlines ordering with thenComparing in createQueue method:

iexec-core/src/main/java/com/iexec/core/task/update/TaskUpdateRequestManager.java

Line 162 in d4a16b6

final Comparator<Task> comparator = Comparator.comparing(task -> task.getCurrentStatus().ordinal(), Comparator.reverseOrder());

[jeremyjams]

Why sorting them by status instead of contribution deadline?

[mcornaton]

When we work with bag of tasks, every task has the same contributionDeadline... But some of them may be more advanced in their statuses, so they have priority to be updated.
However, I agree with you on the fact that sorting tasks by contribution deadline may be great in addition to status ordering and that's something that should not be too hard to implement. See #479 (comment).

[jeremyjams]

Why not something like:

    @Scheduled(fixedDelay = 1000)
    public void consumeAndNotify() {
        if (consumer == null){
            log.warn("Waiting for consumer before consuming [queueSize:{}]", queue.size());
            return;
        }

        while (!Thread.currentThread().isInterrupted()){
            waitForTaskUpdateRequest();
        }
    }

    private void waitForTaskUpdateRequest() {
        log.info("Waiting requests from publisher [queueSize:{}]", queue.size());
        try {
            if (taskUpdateExecutor.getActiveCount() < TASK_UPDATE_THREADS_POOL_SIZE){
                Task task = queue.take();
                CompletableFuture.runAsync(() -> {
                    synchronized (locks.computeIfAbsent(task.getChainTaskId(), key -> new Object())){ // require one update on a same task at a time
                        consumer.onTaskUpdateRequest(task.getChainTaskId()); // synchronously update task
                    }
                }, taskUpdateExecutor);
            }
        } catch (InterruptedException e) {
            log.error("The unexpected happened", e);
            Thread.currentThread().interrupt();
        }
    }

?

[mcornaton]

@jeremyjams: If you already have TASK_UPDATE_THREADS_POOL_SIZE updating threads, you'll trigger a lot of calls to waitForTaskUpdateRequest and load the Scheduler... Which is what I'm trying to avoid there.

[jbern0rd]

Implementation to discuss:

• publishRequest changes to accept a chainTaskId as a parameter. This id is written in a Queue/List, ...
• We use a method to poll on the list (wait on empty list, work on requests)
• We add in taskRepository a method List<Task> findByChainTaskId(List<String ids, Sort sort)
• The new method is used to directly fetch the tasks with the correct order
• The requests update are handled by following the order of tasks returned by the new method
• chainTaskIds from handled tasks are removed from the list

This may allow us to remove some bottlenecks and scheduling problems

[jeremyjams]

streamOfTaskUpdate()
?

[mcornaton]

I don't really like of. The idea here is to tell that the elements in the queue are streamed as TaskUpdates. I don't think streamOfTaskUpdate carries the same meaning: it tells this method returns a stream of TaskUpdates but does not say anything about where these objects come from.

[jeremyjams]

Well-done :)

[jbern0rd]

Thanks for this 👍