Poller fixes

[JamesWrigley]

See commit messages for details.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 89.01099% with 10 lines in your changes missing coverage. Please review.
✅ Project coverage is 91.42%. Comparing base (04fcdc1) to head (6a8e094).
⚠️ Report is 10 commits behind head on master.

Files with missing lines	Patch %	Lines
src/poller.jl	92.77%	6 Missing ⚠️
src/socket.jl	50.00%	4 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##           master     #266      +/-   ##
==========================================
- Coverage   92.50%   91.42%   -1.08%     
==========================================
  Files          11       11              
  Lines         560      548      -12     
==========================================
- Hits          518      501      -17     
- Misses         42       47       +5     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[halleysfifthinc]

I think it would make more sense to handle (i.e. clear) the WAKEUP event immediately? The wakeup doesn't need to exist/persist except to wakeup from the wait/zmq_poll loop and then continue here (and does need to be cleared before the next wait). The later control flow should all still be correct whether or not the poller remains open or closed?

• Poller open: poller.channel is still open (loop continues) and we wait at waiter_wait(poller.barrier) to rearm.
• Poller closed: poller.channel is closed (loop exits)

Suggested change

	pollfd = getfield(item.socket, :pollfd) # don't need to check if socket was closed because that would be caught by the socket error check just above
	pollfd.watcher.events &= ~WAKEUP
	continue

[JamesWrigley]

I had a go but it was quite tricky because of how wait() and close() wakeup all the sockets regardless, and clearing it directly before the polling calls is a race condition. Clearing it in the same task that does the wakeups is the simplest way I think.

[halleysfifthinc]

So my suggestion was definitely wrong, but I think the current state of this PR might be treating a symptom and not the cause?

wait(::FDWatcher) always clears watched for events (line 644 of FileWatching.jl). Reproducer is:

using ZMQ
pull1 = ZMQ.Socket(ZMQ.PULL)
bind(pull1, "inproc://pull1")

notify(pull1, ZMQ.WAKEUP)

wait(pull1) # returns WAKEUP and clears the event in the fdwatcher
getfield(pull1, :pollfd).watcher.events == 0
wait(pull1) # would block because WAKEUP was cleared

So the presence of any leftover WAKEUP's means that some sockets are being notified/woken after they've already synchronized. I played around with this a bit and could not figure out the right logic to avoid the redundant WAKEUPs.

In any case, I noticed from the FileWatching source that we should make sure to grab the fdwatcher.notify lock before clearing events:

@lock pollfd.watcher.notify pollfd.watcher.events &= ~WAKEUP

[JamesWrigley]

wait(::FDWatcher) always clears watched for events (line 644 of FileWatching.jl). Reproducer is:

using ZMQ
pull1 = ZMQ.Socket(ZMQ.PULL)
bind(pull1, "inproc://pull1")

notify(pull1, ZMQ.WAKEUP)

wait(pull1) # returns WAKEUP and clears the event in the fdwatcher
getfield(pull1, :pollfd).watcher.events == 0
wait(pull1) # would block because WAKEUP was cleared

I don't think that line quite explains the behaviour though: https://github.com/JuliaLang/julia/blob/5735163b340af5047919b31e3857bf8b866731c6/stdlib/FileWatching/src/FileWatching.jl#L644
It's only masking the flags that are being watched, so the effect should be that only UV_READABLE is masked. So I'm not sure how the wakeup flag is getting masked too in your MWE 🤔 Will need to grok the code.

So the presence of any leftover WAKEUP's means that some sockets are being notified/woken after they've already synchronized. I played around with this a bit and could not figure out the right logic to avoid the redundant WAKEUPs.

Yeah that's done explicitly by both wait() and close() to ensure that all of the waiters syncronize at the barrier, even if they were already woken up by a readable/writable event.

In any case, I noticed from the FileWatching source that we should make sure to grab the fdwatcher.notify lock before clearing events:

@lock pollfd.watcher.notify pollfd.watcher.events &= ~WAKEUP

I considered that but figured it wasn't necessary since an invariant is that at those points there will be no other task acting on the socket or its watcher. But I'll add it, can't hurt anyway 👍

[halleysfifthinc]

It's only masking the flags that are being watched, so the effect should be that only UV_READABLE is masked.

Don't forget that all sockets now also include the WAKEUP flag in their mask:

ZMQ.jl/src/socket.jl

Line 25 in 04fcdc1

setfield!(socket, :pollfd, FDWatcher(fd(socket), FDEvent(UV_READABLE | WAKEUP)))

[JamesWrigley]

Ah yes you're right, it's the wakeup event that's being caught rather than a readable event.

[halleysfifthinc]

👀 I'll take a look at this in the next several days.

[JamesWrigley]

I'm fairly confident in the approach now so I'll probably merge this in a week or so.

[halleysfifthinc]

Looks good for the most part. My only general concern is that failed watcher tasks are not caught/surfaced any more, and I realized that waiting on the watcher tasks to end in close(::Poller) would rethrow errors from failed watchers. I think a more expected API would be that wait throws if a watcher has failed (and triggers a close?), but close doesn't/won't error (either for failed tasks or timed out waiting for tasks to end). A Poller can't be reopened after a close, and neither error would be actionable for users (besides opening an issue). Thoughts?

[halleysfifthinc]

I'm concerned that this early-return favors earlier sockets, where later sockets will only be returned as ready when the previous sockets have been read/written. As a user, I'd find this somewhat surprising. A round-robin approach (always begin the watcher iteration loop from the next socket of the last returned result) would mitigate this.