# \[ANN\] PriorityChannels.jl
**URL:** https://discourse.julialang.org/t/ann-prioritychannels-jl/19234
**Category:** Package Announcements
**Tags:** task, channel
**Created:** [January 3, 2019, 2:42pm UTC](https://discourse.julialang.org/t/ann-prioritychannels-jl/19234 "2019-01-03T14:42:37Z")
**Posts on this page:** 6
**Page:** 1
### Author:  [@baggepinnen](https://discourse.julialang.org/u/baggepinnen)
#### Post date: [January 3, 2019, 2:42pm UTC](https://discourse.julialang.org/t/ann-prioritychannels-jl/19234/1 "2019-01-03T14:42:37Z")
I’ve created a small package [PriorityChannels.jl](https://github.com/baggepinnen/PriorityChannels.jl) that implements a `PriorityChannel` that mimics [`Base.Channel`](https://docs.julialang.org/en/v1/base/parallel/#Base.Channel), but where each element is associated with a priority. [`take!`](https://docs.julialang.org/en/v1/base/parallel/#Base.take!-Tuple%7BChannel%7D) always returns the highest priority element. Internally, a [heap](https://en.wikipedia.org/wiki/Heap_(data_structure)) is used to keep track of priorities. Usage example in the [readme](https://github.com/baggepinnen/PriorityChannels.jl).
#### Difference between `Channel` and `PriorityChannel`
- `put!(pc, element, priority::Real)` **lower** number indicates a higher priority (default = 0).
- `PriorityChannel` can not be unbuffered (of length 0) and must have a positive length.
- [`take!(pc)`](https://docs.julialang.org/en/v1/base/parallel/#Base.take!-Tuple%7BChannel%7D) returns the highest priority item, `PriorityChannel` thus acts like a [priority queue](https://en.wikipedia.org/wiki/Priority_queue) instead of a FIFO queue like `Channel` does
- Pretty much all other functionality should be the same, including all constructors.
#### Installation
```julia
using Pkg
pkg"add PriorityChannels"
```
Happy New Year, may it be full of wisely prioritized and rewarding tasks!
---
### Author:  [@JesperMartinsson](https://discourse.julialang.org/u/JesperMartinsson)
#### Post date: [September 17, 2021, 7:40am UTC](https://discourse.julialang.org/t/ann-prioritychannels-jl/19234/2 "2021-09-17T07:40:20Z")
Thank you Fredrik for this package! I’ve been looking for this functionality for some time ([RemoteChannel with priority](https://discourse.julialang.org/t/remotechannel-with-priority/18442)) but did not find anything. Do you think this priority functionality can easily be extended to work for [`Distributed.RemoteChannel`](https://docs.julialang.org/en/v1/stdlib/Distributed/#Distributed.RemoteChannel) as well? That would be very handy.
---
### Author:  [@baggepinnen](https://discourse.julialang.org/u/baggepinnen)
#### Post date: [September 17, 2021, 7:59am UTC](https://discourse.julialang.org/t/ann-prioritychannels-jl/19234/3 "2021-09-17T07:59:13Z")
Hello Jesper! This package used some Julia internals that have since changed and this package requires some updates to work even with the standard Channel type. I haven’t looked deeply into RemoteChannel, but if it’s backed by a vector like the regular channel ti should be straightforward.
---
### Author:  [@JesperMartinsson](https://discourse.julialang.org/u/JesperMartinsson)
#### Post date: [September 17, 2021, 9:34am UTC](https://discourse.julialang.org/t/ann-prioritychannels-jl/19234/4 "2021-09-17T09:34:38Z")
Thanks Fredik. It seems to be as easy as for example: `jobs = RemoteChannel(() -> PriorityChannel(32));`. I modified slightly the MWE in Distributed.RemoteChannel and it seems to work in julia 1.7.0-rc1:
```julia
using Distributed
addprocs(4); # add worker processes
@everywhere using PriorityChannels
const jobs = RemoteChannel(() -> PriorityChannel(32));
const results = RemoteChannel(() -> Channel{Tuple}(32));
@everywhere function do_work(jobs, results) # define work function everywhere
while true
job_id = take!(jobs)
println("get $job_id with priority $job_id from jobs")
exec_time = rand()
sleep(exec_time) # simulates elapsed time doing actual work
put!(results, (job_id, exec_time, myid()))
end
end
function make_jobs(n)
for i in 1:n
e = rand(1:500)
println("put job $e with priority $e in jobs")
put!(jobs, e, e)
end
end
n = 12;
# Need to sync before doing work to keep the order
@sync make_jobs(n); # feed the jobs channel with "n" jobs
for p in workers() # start tasks on the workers to process requests in parallel
sleep(0.1) # needed a small sleep to sync the println in remote_do()
remote_do(do_work, p, jobs, results)
end
# Let them finish
sleep(3)
@elapsed while n > 0 # print out results
job_id, exec_time, where = take!(results)
println("$job_id finished in $(round(exec_time; digits=2)) seconds on worker $where")
global n = n - 1
end
rmprocs.(procs())
```
---
### Author:  [@JesperMartinsson](https://discourse.julialang.org/u/JesperMartinsson)
#### Post date: [September 17, 2021, 10:36pm UTC](https://discourse.julialang.org/t/ann-prioritychannels-jl/19234/5 "2021-09-17T22:36:36Z")
You are correct! Something is a bit fishy and not functioning properly. Seems like `put!(jobs,e,e)` hangs if `jobs` is completely emptied by the workers and I later try to add more jobs. If `jobs` are not emptied completely then adding more functions as expected. If I replace `PriorityChannel` with a `Channel` (i.e. the line `const jobs = RemoteChannel(() -> Channel(Inf))`) then workers can empty it completely and I can add more jobs with `put!(jobs,e,e)` as we go.
---
### Author:  [@pitsianis](https://discourse.julialang.org/u/pitsianis)
#### Post date: [September 17, 2023, 6:49pm UTC](https://discourse.julialang.org/t/ann-prioritychannels-jl/19234/6 "2023-09-17T18:49:04Z")
The functionality promised by this package is very useful. Does anybody know how to update it to work with the latest version of Julia?
Alternatively, can somebody who knows the Julia internals propose a workaround for `Base.register_taskdone_hook` that seems to be the point of incompatibility?
Alternatively, is there any other package that provides this functionality, a channel where the elements are associated with a priority value and `get!`'s always return the highest priority?
P.S. The original author @baggepinnen should update the package dependencies to restrict to the latest version of Julia that supports it.