Add priority hint for CoroutineDispatchers

[khovanskiy]

It will be useful for the tasks which depend on the quality of service. For example, we have N abstract clients using the common thread pool via CoroutineDispatcher. In rare moments a few clients have tasks with high priority to execution and they should be computed as soon as possible.
Note: If we create multiple pools for each priority, the situation will turn out that if there are no tasks with a certain priority, some pools will be idle while others may be overloaded.

I guess, that the problem can be solved like that:

1. Create new CoroutineContextElement (for example, CoroutinePriority);
2. If CoroutinePriority is provided by CoroutineContext, wrap a dispatched task into an object which can be compared by taken priority and put it into the executor.

The new context element can thus be used:

launch(Dispatchers.Default + CoroutinePriority(42)) {
    ...
}

The element behaves like a hint. It works only in cases when the dispatcher created from custom thread pool with PriorityQueue and the priority is provided by context, so it produces no overhead in the other cases.

Unfortunately, the most of classes related with Dispatchers are internal in the coroutines package, so it isn't possible to create this logic outside the coroutines library without pain. The another possible solution is to provide the extended API.

[elizarov]

This CoroutinePriority would be helpful if we had an executor that is based on PriorityQueue, but we don't have one. If you create a PriorityQueue-based executor in your project, then you can also define this CoroutinePriority context element in your project. Implementations of CoroutineDispatcher have access to CoroutineContext and can look up any elements when making their dispatching decisions.

I don't quite see the need to have it out-of-the-box in kotlinx.coroutines library. What value would it add?

[fvasco]

Hi @khovanskiy,

If we create multiple pools for each priority, the situation will turn out that if there are no tasks with a certain priority, some pools will be idle while others may be overloaded.

Why creating an executor for "high priority tasks" isn't a good solution?

[khovanskiy]

@elizarov,

Implementations of CoroutineDispatcher have access to CoroutineContext and can look up any elements when making their dispatching decisions.

I agree. At first, I carefully investigated the implementation of ExecutorCoroutineDispatcherBase to create a custom Dispatcher, but the base has many internal fallbacks such as DefaultExecutor and EventLoop usage, that can't be directly used outside library. It turns out that the only option to maintain the standard behavior is to copy most of the library. So as I wrote previously the another possible solution is to provide the extended API, which helps to create custom Dispatchers via extending standard ones. If ExecutorCoroutineDispatcherBase is made public, it will solve this issue, but API can be probably enhanced better.

@fvasco,

Why creating an executor for "high priority tasks" isn't a good solution?

It is a possible solution, but not the best one. If I have no another solution, I will have to do so. The problem occurs when the application works as a microservice (nanoservice) or a mobile app and has too few resources. Each thread pool produces overhead and in the case when we have a special pool for high-priority tasks, we doesn't effectively utilize resources. Also the high-priority is a particular case. In the real application we can have multiple of them like: minor, normal, major, critical etc, and the efficiency will decrease even more.

[elizarov]

You don't really need to maintain all the internals. Look, a dispatcher implementation can be as simple as that:

class MyDispatcher : CoroutineDispatcher() {
    private val executor = Executors.newCachedThreadPool() // or anything else

    override fun dispatch(context: CoroutineContext, block: Runnable) =
        executor.execute(block)
}

[khovanskiy]

@elizarov,
I thought, that custom Dispatcher should implement Delay interface to support schedule functions like delay and withTimeout, but it even works with your simple example using DefaultDelay when custom Dispatcher doesn't implement it.
Thank you for the explanation!

[fvasco]

Coroutines aren't preemptive, I fear that too many categories become too hard to tuning. It is not possible to define what execute before, instead it is only possible define what execute later (possibly never on a high load).

I feel we should delegate this issue to thread's priority, it is not cheap, I agree with you, but it works on any load and its misconfiguration is not a pain.

[Richie94]

Hey, Im having the same request and wanted to have on one hand some validation that this is how it should be written and on the other hand to have some online example to copy.

class PriorityExecutor {

    private val priorityDispatcher = PriorityDispatcher()

    internal class PriorityDispatcher : CoroutineDispatcher() {

        private val priorityExecutor = ThreadPoolExecutor(
            1, 1, 0L, TimeUnit.MILLISECONDS, PriorityBlockingQueue(100000),
            CustomizableThreadFactory("importRunner-serializer")
        )

        override fun dispatch(context: CoroutineContext, block: Runnable) {
            val priority = context[CoroutinePriority]?.priority ?: 0
            val executable = RunWithPriorityFuture(RunWithPriority(block, priority))
            priorityExecutor.execute(executable)
        }

    }

    internal data class CoroutinePriority(
        val priority: Int
    ) : AbstractCoroutineContextElement(CoroutinePriority) {
        companion object Key : CoroutineContext.Key<CoroutinePriority>

        override fun toString(): String = "CoroutinePriority($priority)"
    }

    internal class RunWithPriority(val runnable: Runnable, val priority: Int) : Runnable {

        override fun run() {
            runnable.run()
        }
    }

    internal class RunWithPriorityFuture(private val runWithPriority: RunWithPriority) :
        FutureTask<RunWithPriorityFuture>(runWithPriority.runnable, null), Comparable<RunWithPriorityFuture> {

        override fun compareTo(other: RunWithPriorityFuture): Int {
            return other.runWithPriority.priority - runWithPriority.priority
        }

    }

    fun submitAction(priority: Int, runnable: suspend () -> Unit): Job {
        return CoroutineScope(priorityDispatcher).launch(CoroutinePriority(priority)) {
            runnable.invoke()
        }
    }
}

is this the way @elizarov you would do it? The priorityqueue/threadexecutor under the hood is taken from java examples, I just want to be able to also run suspending functions when I call it