Julia: Passing data between coroutines (tasks)

Question

I have tried to get two tasks (coroutines) to cooperate, but to no avail. Below is what I have. I never see the printlns from either source or sink functions, and the wait() call appears to hang forever. I have also tried making the p (source) task a global variable, instead of passing it to sink() as an argument, but that does not seem to work either (even when I declare it inside sinke() as a global.

This is the kind of thing that I discovered I can do very easily with channels and goroutines in Go.

I have also experimented with calling yieldto() inside the source() and sink() functions, but I still seem to end in deadlock.

Anyone with an example of multiple tasks sharing data in any way? Ideally, I would have a pipeline, or long chain of tasks.

Thanks in advance for your help.

println("Hello")

function source()
  println("source start")
  produce("start")
  produce("stop")
end

function sink(p::Task)
  println("sink start")
  println(consume(p))
  println(consume(p))
end


a = Task( source )
b = Task( () -> sink(a)   )

wait(b)
wait(a)

println("Goodbye")

Answer 1

In Julia a creating a task does not automatically schedule that task. The wait function doesn't schedule either so you end up with a deadlock. This is a big difference from Go where the go statement takes care of all the scheduling for you. In julia you have to do a little more work. Specifically use the @sync and @async macros to make it easier.

@sync begin # @sync will wait for all the contained tasks to terminate.
   a = @async source() # @async will create and schedule a task for you automatically
   @async sink(a)
 end

You'll notice though that this doesn't terminate either. All of the prints happen but the task does not terminate. The cause is @sync waiting for task a to finish but task a has not been scheduled. Add a final consume or schedule to your sink function to force task a to get scheduled one final time so it can terminate. Or better yet use a for loop on the task so you always exhaust it.

println("Hello")

function source()
  println("source start")
  produce("start")
  produce("stop")
  println("source end")
end

function sink(p::Task)
  println("sink start")
  for s in p
    println(s)
  end
  println("sink end")
end

@sync begin
   a = @async source()
   @async sink(a)
 end

println("Goodbye")

Julia's tasks are cooperatively scheduled which basically means you have to ensure each task gets scheduled by yourself. The runtime isn't going to do it for you. Luckily the @sync and @async macros do most of this for you.