https://jacobzelko.com/01082023043553-julia-async-workflow/

[giscus[bot]]

https://jacobzelko.com/01082023043553-julia-async-workflow/

A thorough tutorial on how to use Julia for asynchronous workflows to do many things, at once!

https://jacobzelko.com/01082023043553-julia-async-workflow/

[Nimrais]

Thanks for the post! There are clear advantages to the proposed workflow.

However, there is no straightforward way to stop watching file changes if one does not need it anymore.

For example, Rocket.jl and Observables.jl (high-level reactive programming modules) have implemented subscribe/unsubscribe concepts.

There is an example of a possible Rocket based implementation:

using Rocket
using FileWatching

file_to_watch = "/Users/mykola/tmp/tmp.txt"

ticks = BehaviorSubject(Int, 1)

actor = (d) -> begin
    watch_file(file_to_watch)
    println("Change Detected $(file_to_watch)!")
    next!(ticks, d+1)
end

subscription_to_file_changes = subscribe!(ticks |> async(1), actor)

for i in 1:15
    println("Hi from the main flow $(i)! I can react on the file changes")
    sleep(2)
end

unsubscribe!(subscription_to_file_changes) # pff! I don't care about the file anymore

for i in 1:5
    println("Hi from the main flow $(i)!")
    sleep(2)
end

During the execution of the code above I add several lines into /Users/mykola/tmp/tmp.txt

Hi from the main flow 1! I can react on the file changes
Hi from the main flow 2! I can react on the file changes
Hi from the main flow 3! I can react on the file changes
Change Detected /Users/mykola/tmp/tmp.txt!
Change Detected /Users/mykola/tmp/tmp.txt!
Hi from the main flow 4! I can react on the file changes
Change Detected /Users/mykola/tmp/tmp.txt!
Hi from the main flow 5! I can react on the file changes
Hi from the main flow 6! I can react on the file changes
Hi from the main flow 7! I can react on the file changes
Hi from the main flow 8! I can react on the file changes
Hi from the main flow 9! I can react on the file changes
Hi from the main flow 10! I can react on the file changes
Hi from the main flow 11! I can react on the file changes
Hi from the main flow 12! I can react on the file changes
Hi from the main flow 13! I can react on the file changes
Hi from the main flow 14! I can react on the file changes
Hi from the main flow 15! I can react on the file changes
Hi from the main flow 1!
Hi from the main flow 2!
Hi from the main flow 3!
Hi from the main flow 4!
Hi from the main flow 5!

[TheCedarPrince]

Hey @Nimrais , this is really fascinating. I had never seen Rocket.jl before and didn't know you could use Observables.jl like that. You are absolutely right as this gets around a deficiency in my approach. Would you be willing to explain a bit further on the example you gave? In particular, what does this mean -- is it generating some sort of reactive asynchronous object?

ticks = BehaviorSubject(Int, 1)

In what follows, what is (d) and what does next! mean?

(d) -> begin
    watch_file(file_to_watch)
    println("Change Detected $(file_to_watch)!")
    next!(ticks, d+1)
end

Thanks for sharing this example -- this is really great.

[Nimrais]

s it generating some sort of reactive asynchronous object?

ticks = BehaviorSubject(Int, 1)

BehaviorSubject(Int, 1) is not an async object itself, in Rocket it stands for an container that send a message to its subscribers when a new value assigned to it.

In the code before I created from ticks with async(1) operator ticks |> async(1): a non-blocking observable. ... |> async(1) creates an async object, ticks |> async(1) will send messages to it's subscribers asynchronously. So ticks |> async(1) is the actual asynchronous object here.

Actually we can simplify the actor: we don't care about the exact value of the message. The obtaining of the message itself is a trigger. So actor can be simplified in the following way:

actor = (_) -> begin
    watch_file(file_to_watch)
    println("Change Detected $(file_to_watch)!")
    next!(ticks, 1)
end

what does next! mean?

How do we assign a value to an instance of BehaviorSubject? For this Rocket.jl has next!. So next(ticks, 1) one card read: assign value 1 to ticks (when next! called on the BehaviorSubject instance).

So let us translate actor logics then: when a message is received, it is watching file for a change watch_file(file_to_watch), and when watch_file(file_to_watch) is ended, it assigns a new value to ticks.

Finally, subscribe!(async_ticks, actor): we are subscribing our actor to receive messages from async_ticks. You can think of it as a while true loop in this context: ticks emits a message into actor, actor waiting for the file change and when it has happened assigns a value to ticks, the assigning push ticks to emit a message into actor, ...

A bit more structured code then:

using Rocket
using FileWatching

file_to_watch = "/Users/mykola/tmp/tmp.txt"

ticks = BehaviorSubject(Int, 1)
async_ticks = ticks |> async(1)

actor = (_) -> begin
      watch_file(file_to_watch)
      println("Change Detected $(file_to_watch)!")
      next!(ticks, 1)
end

subscription_to_file_changes = subscribe!(async_ticks, actor)

for i in 1:15
    println("Hi from the main flow $(i)! I can react on the file changes")
    sleep(2)
end

unsubscribe!(subscription_to_file_changes) # pff! I don't need this file anymore

for i in 1:5
    println("Hi from the main flow $(i)!")
    sleep(2)
end

[TheCedarPrince]

This is really fantastic and makes a lot more sense @Nimrais . Thanks for clarifying and the in-depth explanation! Actually, if you'd be interested, would you want to add a section to this blog post as a PR? I think your approach adds a lot of value to this post and it would be a shame for it to be relegated to the discussion. I'd be happy to include you as a co-author on the post going forward. 😄

P.S. All you would need to do in making a PR is fork the site and add content to the note here: https://github.com/TheCedarPrince/thecedarprince.github.io/blob/main/notes/01082023043553-julia-async-workflow.md

[josephsdavid]

Wow this is really cool! I need to spend some time to figure out a similar workflow for tests (while watching affected source code) using ReTest :)