Properly log progress messages #326
Conversation
|
This doesn't address the same issues as #321, but we should double check whether it relieves any pressure there. There should be only some gap between the receive stream and the inputs to the progress tracker. EDIT: This is not exactly true. Post-propagation pointstamps (e.g. from subgraphs that exchange their own pointstamps) will not be exchanged, and annoying work would need to be done to reconstruct them just from the exchanged updates. |
|
Thinking out loud, although these logging events are informative, they aren't exactly sufficient to track the frontier of the system (which would be a desirable goal). For example, operators are often created with an initial capability, which their logic then drops and workers confirm by sending progress messages around. So we see the drops, but not the initial capabilities. It would be nice to directly log the dataflow frontier in the progress tracker, which is a bit more in line with #321 (though, not something that #321 produces as output). Afaict, #321 produces the raw pointstamp updates in progress tracking, which is possibly more immediately helpful than the progress messages for determining the state of capabilities in the system. |
| let mut internal = Vec::with_capacity(changes.len()); | ||
|
|
||
| // TODO: Reconsider `String` type or perhaps re-use allocation. | ||
| for ((location, time), diff) in recv_changes.iter() { |
There was a problem hiding this comment.
I've done this locally as well to track down a potential race, and ended up just changing messages and internal to a single vector of changes, as this filtering can be easily done at the receiver. Any reason you prefer to keep them separate?
There was a problem hiding this comment.
Oh, only that it was written that way to begin I think, and I wanted to change minimal things. Did you make the change because this was error-prone? Historically they were separate because that's how the data were presented (separate messages and internal).
There was a problem hiding this comment.
I think also historically vectors of enums where kept as multiple demux'd vectors to minimize branching in the middle of hot loops (back when the time to update an operator was sub-microsecond. I think that time may have drifted up enough that this sort of attention to detail is less critical).
|
|
||
| /// Expose the internal vector of updates. | ||
| pub fn unstable_internal_updates(&self) -> &Vec<(T, i64)> { &self.updates } | ||
| pub fn unstable_internal_updates(&self) -> &[(T, i64)] { &self.updates[..] } |
There was a problem hiding this comment.
I'm not sure what this change was about, but I think this reduces functionality (we can no longer check the internal capacity). I bet I thought I was just tidying, but I think I'll skip doing this.
|
Closing in favor of #352, which is just a cleaned up version of this PR. |
2 participants
Log progress messages, converting timestamps to strings. This puts more allocation behind each of these messages (from an empty vector to a vector of strings) but there doesn't seem to be much to do about it if we want to see the actual progress contents moving around.
The receive side has the potentially to be much spammier than the send side, as this is a broadcast. Depending on the volume we may want to dial that down and just presume that sent messages are not lost in transit (we would lose the timing information about when they are received though). For the moment, we log lots of things.
cc: @utaal @saradecova