Conversation
mythical-fred
left a comment
There was a problem hiding this comment.
Nice work. Two small nits inline; nothing blocking. Commit messages are exemplary — the per-commit "why" explanations are exactly the standard we want.
| impl MessageSize { | ||
| /// Constructs `MessageSize` from a count of `bytes`. | ||
| pub(crate) fn from_bytes(bytes: usize) -> Self { | ||
| if bytes <= 4096 { |
There was a problem hiding this comment.
nit: the 4096 threshold is a magic number. Consider hoisting it to a named const SMALL_MESSAGE_MAX_BYTES: usize = 4096; with a one-line comment on the rationale (page size? empirical? aligned with something on the wire?). Makes it easier to tune later and self-documents intent.
| /// same (micro)step in every worker. This is true for input operators, | ||
| /// which flush as soon as the transaction starts committing, but it is not | ||
| /// necessarily true for other operators. | ||
| InputOnly, |
There was a problem hiding this comment.
The "Limitation" here (operators must flush in the same microstep on every worker) is a real correctness invariant, and ExchangeActivity is now pub — reachable via Sharder::with_activity. It would be easy for someone downstream to slap InputOnly on a non-input operator and get subtle wrong-answer bugs that only surface in multihost.
Two possible mitigations, either would work:
- Restrict
with_activity/ExchangeActivity::InputOnlyvisibility topub(crate)until we actually need it elsewhere. - Add a
debug_assert!somewhere on the commit path that trips when anInputOnlyexchange sees a non-empty message during the commit microstep, so misuse fails loudly in tests instead of silently.
Not a blocker — just a footgun worth defusing while the surface area is still small.
|
These commits were in a prior PR, but I took them out of that PR because they needed performance testing. This means that they might be familiar to some readers. |
mihaibudiu
left a comment
There was a problem hiding this comment.
I read this, but I think @ryzhyk will have to review it as well.
I don't know this part well enough to approve.
| #[derive(Copy, Clone, Debug, PartialEq, Eq, Enum)] | ||
| pub(crate) enum MessageType { | ||
| /// Messages sent via synchronous exchange. | ||
| Synchronous( |
There was a problem hiding this comment.
Is the plan to eventually deprecate synchronous exchange?
| /// as needed. | ||
| pub(crate) async fn receive_all<D>(&self, deserialize: D) -> Vec<T> | ||
| /// | ||
| /// When the data is ready, but before reading it, this method swaps |
| .receive_all(deserialize, Some(&self.start_wait_usecs)) | ||
| .await; | ||
| if let Some(start_wait_usecs) = start_wait_usecs { | ||
| self.total_wait_time.fetch_add( |
There was a problem hiding this comment.
I guess this is where the time is used
| #[track_caller] | ||
| pub fn dyn_shard_pairs( | ||
| &self, | ||
| pub fn shard_pairs( |
There was a problem hiding this comment.
Isn't the "dyn" naming convention supposed to be used here?
Or that applies only to Stream objects?
There was a problem hiding this comment.
I didn't think it was necessary because Stream::dyn_shard is how you get the Sharder object that this method belongs to. I'm happy to change the name if it's helpful to anyone.
ryzhyk
left a comment
There was a problem hiding this comment.
The last commit message says
A similar change is needed for remote waiters; this is left for another
commit.
But it looks to me like it already adds a remote waiter operator as well. What's still missing?
Yeah, that's wrong. It was previously true, but then I wrote another commit that added the remote waiter operator, and then I squashed the commits together without being careful about the commit message. I will fix the commit message. |
We don't need so many special cases because `dyn_shard()` is cached, for the case where there is a sharded version already, and a no-op, for the case where there is only one worker. Signed-off-by: Ben Pfaff <blp@feldera.com>
This implements `dyn_shard_workers_accumulate` in a streaming manner by sharding only to a subset of workers. Signed-off-by: Ben Pfaff <blp@feldera.com>
DBSP has the notion of an asynchronous operator, which is one that it does not evaluate until it indicates that it is ready. Since this notion was introduced, DBSP was changed to use Rust async for its evaluation functions. Rust async is more flexible and easier to use than DBSP asynchronous operators, so this commit changes ExchangeReceiver to use it instead. This is preparation for updating ExchangeReceiver::eval() to work in a way that doesn't fit as well with the DBSP asynchronous operator concept. After this commit, there is only one DBSP asynchronous operator left, which is RebalancingExchangeSender. It could probably use the same treatment, and then afterward we could get rid of the DBSP asynchronous operator concept entirely. Signed-off-by: Ben Pfaff <blp@feldera.com>
…commit. Input operators never exchange any actual data once a circuit's transaction starts to commit, but until now they still incurred the overhead and latency of an exchange. This commit fixes that for at least two input operators (perhaps others need improvements too). The most interesting part of this commit is the new builder for sharding, `Sharder` in `crates/dbsp/src/operator/dynamic/communication/shard.rs`. I don't think we've used this pattern before for constructing operators. It seems appropriate in this case because sharding is now configurable among multiple axes and it was getting unwieldy to add new methods for each combination. Signed-off-by: Ben Pfaff <blp@feldera.com>
Until now, when one host in a DBSP pipeline sends another one a message as part of synchronous or streaming exchange, it sent it over a single host-to-host TCP connection. This means that synchronous exchange has to wait for streaming exchange messages, which increases latency. This commit uses different TCP connections for streaming and synchronous exchange, which helps. In addition, it uses separate TCP connections for small and large synchronous exchange; we speculate that giving small synchronous messages priority can help with broadcast and consensus messages (this is unproven). Signed-off-by: Ben Pfaff <blp@feldera.com>
This should be equivalent code, just shorter. Just noticed this while reading code, there is nothing important here. Signed-off-by: Ben Pfaff <blp@feldera.com>
…erator. Before this commit, the sharded accumulator waited for spines to merge in the ShardedAccumulatorSender operator before it returned a value from its eval() method. This prevented any other work from happening in the receiver and any other part of the circuit downstream from the receiver, because there is an artificial dependency between the sender and receiver created by add_exchange_operator(). One way to solve the problem is to remove that dependency (it is not essential for streaming exchange) but it still makes some sense to have it; it does make sense to receive after sending. Another reason is that ExchangeActivity::InputOnly currently expects all of the workers to flush in the same step, but without the dependency I am not confident without further thought that that would still be the case, since it would become nondeterministic which operator (sender or receiver) would run first. Instead, this commit breaks off the waiting part of ShardedAccumulatorSender into a new operator ShardedAccumulatorWaiter, which just does the wait. It adds a dependency on ShardedAccumulatorReceiver for that operator, so that it runs after everything has been sent and received in a step. The new waiter operator does no work, it only keeps the step from finishing before the spine used for accumulating data has merged to an acceptable number of batches. This way, the sender runs quickly and the wait does not block the receiver. Remote waiters are a different problem. This commit handles remote waiters by adding a single operator to wait for all of them at the very end of a step (by making the operator depend on all the sending operators). This isn't a problem in synchronous exchange (or perhaps it is a different problem) because there it is the receiver, not the sender, that adds the batch to the spine. Signed-off-by: Ben Pfaff <blp@feldera.com>
Signed-off-by: Ben Pfaff <blp@feldera.com>
This series of commits has been benchmarked across a variety of customer-like workloads and achieved between 5% and 50% performance improvement versus main for 2-host deployments.
Thanks to @venkatamudili-fel for running benchmarks on this over the last 2 weeks.