Observability support

[JamesWrigley]

Things that we could do:

• Keep counters that track communication statistics between workers, like the number of RPC calls and the amount of data sent/received.
• Expose a hook API so packages can register functions that will be called on sending/receiving data. Not sure about the performance implications of this.
• Read-only API to check the status of a worker, the time of last contact, and state of worker-to-worker connections.

Use cases:

• Tracking network usage between workers on HPC clusters
• Better support for profiling tools like Extrae (Roadmap #1 (comment))

[rayegun]

This is not really a comment on the high level API, but in terms of the low level communication middleware it is easy to do either counters or hooks but not both. Of course you can build the former on top of the latter without much issue but I thought I would make note of that from a performance POV. Ideally the hooks API can support both high-level observability and low-latency use-cases

[JamesWrigley]

Why is it difficult to do both? I could imagine a naive implementation where e.g. the message handler task would be incrementing some counters and then calling hooks for each message, and a background task could periodically sync the per-worker counter state to the master.

[rayegun]

To be clear I'm not saying this is impossible or even super difficult I was just mentioning the impedance mismatch between the transports like Libfabrics and the high level interface here. Libfabrics for instance supports incrementing a counter or pushing an entry into the completion queue (for implementing callbacks for instance). You can just use a callback to do both of course, but that has performance implications and adding hooks to all communication operations can have a big impact on latency.

A better design IMO, and how I aim to support completion in upcoming low level communication runtimes (look out for the release of this around JuliaCon :)) is an extensible completion mechanism where an application submits a communication call and specifies how they want to complete it: counter, callback, queue, MPI style Request object, or graph completion. Then when any thread progresses the endpoint these objects are (atomically) signaled as complete. When the caller resumes it can then also do whatever post-processing it wants after waiting on the completion object.

Essentially it is much better IMO for a communication call to opt-in to the functionality it wants rather than looping over an arbitrary number of callbacks for each completion.

Perhaps the other real issue I should mention is thread safety of hooks. For now I'll just mention that we can get very good performance scaling by creating many communication endpoints within a single process and callback hooks must be thread and task migration safe for that reason.

[jpsamaroo]

@rayegun any chance you can show an example of this sort of extensible completion mechanism, for the sake of discussion? I'm having trouble grasping the idea from words alone 😄

I personally think James' point 3 is immediately and easily doable and should be done independently of the rest (and may not be a great target resource for callbacks anyway).

[rayegun]

Hmm, looks like I need to work on my clarity here!

I don't have the packages ready for publishing. But at a high level:

e = Endpoint(capabilities = READ, WRITE, SEND, RECV)

# thread 1
counter_1 = CompletionCounter(e)
send!(e, counter_1, payload)
write!(e, counter_1, payload)

# thread 2
foo_callback() = println("silly callback")
read!(e, foo_callback(), output_payload)
write!(e, foo_callback(), output_payload)

# thread 3
progress(e) 
# prints 
# silly callback 
# silly callback

# thread 1
wait(counter_1,  2) # progress(e) if counter is not equal to 2, but it should be if thread 3 already finished progressing.

# thread 2
# nothing to do, but perhaps foo_callback could push to a queue and I could check it here

In the end counter_1 will have a value of 2, and the callback foo_callback will have been called twice. This enables a few things:

1. If we're in a hot part of the application we can use counters for completion with low latency without preventing other libraries from running simultaneously with strong observability.
2. Scales better than registering a hook with some centralized mechanism
3. Reduces likelihood that someone's funky high contention or incorrectly implemented hook will block and slow down everyone else
4. Other completion mechanisms like a request that you wait on, a queue, or a graph of communication calls

Again, this is much lower level stuff than Distributed. But avoiding global hooks at the Distributed level means we can build Distributed on top of Libfabrics and UCX without causing any issues.

[JamesWrigley]

Ok I think I glossed over this part in your first comment:

low level communication middleware

If I understand you correctly, you're saying that by registering hooks with the underlying communication layer like UCX we can avoid calling hooks directly in DistributedNext and register them with the communication layer instead, which will call them for us e.g. whenever sending/receiving data? And the advantage of an asynchronous hook API (the 'completion' part) is allowing the user-defined hooks to run with minimal communication overhead?

[JamesWrigley]

Or maybe I misunderstood completely, are you saying that we could have CompletionCounter for all objects/operations and tell users to wait(counter, 1) on them to know when e.g. a remotecall has been initiated?

[rayegun]

Each completion (the completion of a send, write, read, or recv) can trigger a single callback. This could be used to do anything and trigger an arbitrary number of other callbacks, but in general we want to use it for something very very lightweight, atomic, and fast. Hooks are the opposite of that.

My issue with the use of hooks is that they are global, so I'm really just trying to motivate avoiding global hooks across the entire stack. We can handle hooks within our UCX and Libfabrics callbacks / progress engine but that would slow everyone down (and those hooks must be carefully thread-safe and ideally lock-free) so we shouldn't do that.

But if we put the hook functionality a layer up in DistributedNext.jl we cause the same issue just at a higher level. If we run hooks on every call to DistributedNext.remotecall then we are necessarily slowing everyone down. Instead I would prefer we allow users (or packages) to control what happens upon completion of individual operations. The performance implications are a lot better.

[JamesWrigley]

Sorry, I'm still not getting it 😅 Isn't this sentence:

Each completion (the completion of a send, write, read, or recv) can trigger a single callback. This could be used to do anything and trigger an arbitrary number of other callbacks, but in general we want to use it for something very very lightweight, atomic, and fast. Hooks are the opposite of that.

A contradiction of this sentence?

Instead I would prefer we allow users (or packages) to control what happens upon completion of individual operations.

If not, what does that sentence mean? Isn't controlling what happens upon completion the callback thing you mentioned? Or is that a separate mechanism?

I'm also not sure if putting hooks at the communication layer level is what we want. We don't just want to monitor all network traffic, ideally we would be able to monitor traffic e.g. per RPC call type (remotecall, remotecall_fetch, etc). Would that be possible with a completion design?

[rayegun]

It's a mild contradiction. UCX and Libfabrics interfaces will have the following completion mechanisms: counter, queue, callback, request, graph. The callback there is indeed a somewhat dangerous interface, but necessary for full generality. It should only really be used by advanced users (for instance by DistributedNext.jl to implement functionality it really needs like refcounting). DistributedNext could be a responsible user of this, but if it runs an arbitrary number of hooks in this callback that's probably bad.

The issue with hooks as I understood the OP is that a user would register a hook with DistributedNext, and then DistributedNext would run all of it's hooks using the callback completion mechanism. This does indeed limit the hooks to be run only on the communication calls created by DistributedNext but it's still burdensome and if we want DistributedNext to be fairly standard then those hooks still have abig impact.

Sorry I did a poor job of motivating this, in the end I just want to say hooks are scary global behavior. At some point we should hop on a call and I can show the current API for Libfabrics.jl and UCX.jl and we can have a back and forth about what DistributedNext.jl needs and what the low level packages can provide and meet in the middle. But that'll have to wait until the end of this semester. IMO it would be nice to have different completion mechanisms available to the user even at a high level, but that is somewhat orthogonal to whether DistributedNext will have a hooks API.

[JamesWrigley]

Sounds good, and I'm very interested to see the work on UCX.jl and friends :) That's been on my radar for a while.