First pass at a Go-style select statement

[malmaud]

Progress:

• Blocking macro variant
• Non-blocking macro variant
• Non-macro form
• Proper error handling
• General cleanup

Proposed syntax:

    @select begin
        if c1 |> x
            response = "Got $x from c1"
        elseif c2
            response = "Got a message from c2"
        elseif c3 <| :write_test
            response = "Wrote to c3"
        end
    end

cf Go select construct grammar

[tkelman]

add this to test/choosetests.jl, and add license header

[tkelman]

• docs

[ViralBShah]

cc @amitmurthy

[amitmurthy]

Can this become generic enough to be able to wait on any waitable type ? Tasks, sleep (to implement a timeout), condition variables, etc?

[malmaud]

That seems like a good goal. But in

@select if condition |> value
  @show value
end

what should value be when condition isn't a channel? Perhaps the return value of wait(condition) that led to this branch of the select statement being run. Seems a little unfortunate that value=take!(condition) if isa(condition, AbstractChannel) and value=wait(condition) otherwise, but it's at least less unfortunate than a whole different select-like function for non-channel conditions. '

Or that store-value select case type could be disallowed for non-channels, with only

@select if condition
println("$condition finished first)"
end

working. I think at this point, the only type of condition where you would typically want the return value of wait is when condition is a Task, but the return value of the task could just be accessed through condition.result inside the branch anyways.

[malmaud]

All waitables are supported now. c |> x now resolves to x=take!(c) if isa(c,AbstractChannel), x=c.result if isa(c,Task), and nothing otherwise.

[amitmurthy]

Nice! I was wondering if the implementation would have been simpler if we had a way to kill a task. In that case we would not need the extra lock in Channel and pair of (waiters, evaluators) - the "winning" evaluator could just kill the other tasks....

[amitmurthy]

A couple of issues I just realized.

1. There is a possible race condition between the waiter task triggering the notification channel and the evaluator block being run. Another task could execute a put!/take! in between. I think you will need to combine the two blocks anyways. It may just be a good idea to implement kill(::Task) and simplify the implementation.

2. winner_ch has to be at least as long as the number of waiters. Else some of the waiter tasks may never exit.

[malmaud]

My intention was the new lock field in channels would deal with 1)- nothing can take!/(put!) from the winning channel between the waiter's call to wait finishing and the executor running. You don't think that works as intended? I didn't finish writing comments yet, which should explain that logic.

As a bonus to that approach, channels will have to become lockable to be threadsafe anyways once the multithreaded branch becomes used, so it seemed a way of killing two birds with one stone.

I talked with @vtjnash a while ago about the possibility of a task-killing function- he seemed reluctant about that idea, but I can't remember why now. Maybe he can chime in here. It does seem like that's the simplest approach unless there is a deep issue with killing tasks I'm not aware of.

Definitely you're right about 2).

[amitmurthy]

Scenario

Task 1 -> @select task with one of the conditions being a take! on channel c
Task 2 -> @schedule waiter task on channel c
Task 3 -> Some other task also blocking on a take! on c

Task 4 puts data into c

Both Task 2 and Task 3 are notified

Task 2 is is run and it writes into winner_ch

edit: Now before Task 1 is run, Task 3 gets executed which removes data from c

Task 1 now tries a take! on c which blocks

Cleaning up the waiting tasks immediately (point 2 above) will not be easy since they are are all waiting on the actual objects. Even with a longer winner_ch, tasks will be hanging around till the wait conditions are triggered.

We need a kill task - some discussion here - #6283

[vtjnash]

the presumed need to kill a task is probably a flaw in the API design here. it's hard to make specific comments without a more concrete proposal (aka Julep) to reference though.

this sounds like it may be starting down the path of implementing the waitq idea from https://github.com/JuliaLang/julia/pull/7275/files#diff-121162110854f09c3be5b39209646009R220, so dropping this reference in as a cross link.

[malmaud]

But doen't the lock in https://github.com/JuliaLang/julia/pull/13661/files#diff-ab1c387bfae5c04f18ba106ec410c7f8R49 block Task 4 from removing data, given https://github.com/JuliaLang/julia/pull/13661/files#diff-ab1c387bfae5c04f18ba106ec410c7f8R214 acquired the lock earlier? Task 4 gets put back on the wait queue while it waits for the mutex, meanwhile Task 3 is run successfully, unlocks the mutex, and Task 4 gets scheduled again.

Although that does reveal a different problem with the code as written, since once the mutex is unlocked and Task 4 is rescheduled will at that point try to take without checking again that data is still available, even though the channel is empty. The code could be changed to cope with that though.

[amitmurthy]

I meant Task 3 takes the data which was meant to be taken by Task 1 (the one with eval blocks in @select)

[amitmurthy]

Also I have not clearly understood the need for the extra lock - is it only for the multi-threading case? If so, we should drop it at this point and only consider it when tasks themselves become schedulable on different threads.

[malmaud]

That lock is exactly intended to deal with the race condition you're bringing up.

Let me explain it more clearly what I was going for:

The waiter looks like

wait(channel)
lock(channel)
put!(winner_ch, 1)
unlock(chanel)

So once wait(channel) executes, there is data in the channel. Then lock(channel) is called. put! is then called, queueing the waiter and scheduling channel's listeners, which includes at a minimum the evaluator. (There is a flaw here that needs to be fixed - this code does assume that the evaluator will be scheduled for the waiter is scheduled again.)

Now say at some point before the evaluator can call take!(channel, uselock=false), another task calls take!(channel, uselock=true). This will block on https://github.com/JuliaLang/julia/pull/13661/files#diff-ab1c387bfae5c04f18ba106ec410c7f8R103, since the uselock=true argument causes take! to try to acquire the channel's mutex. So now that task is put back on the scheduler, the evaluator is eventually scheduled and runs take!(channel, uselock=false) successfully, because the uselock=false causes take! to not to try acquire the mutex and hence block.

[amitmurthy]

AFAIK put! will not yield to other tasks immediately. The entire block of

wait(channel)
lock(channel)
put!(winner_ch, 1)
unlock(chanel)

will be executed without yielding to other tasks in between. Notified tasks (due to the put!) are just added to the scheduler's Q.

[malmaud]

Oh, does it not yield? My bad. I guess it would need to explicitly yield and have some logic to ensure that it stays blocked until the evaluator is run.

I looked at the waitq code - we could implement the same idea. Make just one task per condition, instead of a separate waiter and executor. Each task is listening for a special shutdown exception in a catch block. The winner raises that exception on all its rival tasks as soon as its wait statement finishes.

That is probably not super from a performance perspective, but it seems simple and robust.

[malmaud]

OK, here is take 2. A cleaner rewrite of the implementation using the task-killing strategy. Still needs docs, but otherwise I think it's ready.

[vtjnash]

still needs docs, and preferably a implementation spec to describe what you are attempting to accomplish here.

[malmaud]

were you thinking that spec would be via a comment block at the top of the new code, or something like an external markdown document that I link to?

[malmaud]

I could create a Julep issue and link it to this PR.

[vtjnash]

Just doing some cleanup of PRs that might be stale