Do not hold on to block data in the plot worker

[cfoust]

User-Facing Changes
Drastically reduce the amount of memory required for plotting.

Description
In the beginning, I moved the plot pipeline to a Web Worker in order to reduce the burden that dataset creation and downsampling put on the main thread. This was a big change, so to reduce the number of variables I kept the logic for aggregating messages the same. As a result, we had two copies of the message data used for plotting: one in the main thread and another in the worker. This approach cost a lot of memory, but interacting with plots was much faster.

This PR finally closes the loop and makes it so that the main thread sends the messages that plots need on demand; the worker only holds on to plot datasets and not the messages necessary to produce them. This greatly reduces the memory usage of plots, with the downside that the relevant raw messages must be transmitted to the worker in full when plot parameters change.

This costs a fair bit of complexity, so to make this easier to test and reason about I broke out the logic for sending messages to clients into a standalone state machine that's used by useDatasets and added a suite of tests.

Known issues
There is still a timing issue affecting plot stories that I am investigating. Update: fixed!

[cfoust]

I fixed the issue with current data. We now retransmit it to a specific client when plot params change. Right now we transmit all current data subscribed to by plots, not just what the client actually needs, so it's a little wasteful but I'm still polishing.

[defunctzombie]

relevant raw messages must be transmitted to the worker in full when plot parameters change.

What plot parameter changes cause this to happen?

[defunctzombie]

Not a fan of all these globals - why do we have globals here?

[cfoust]

It's module-level state that exists outside of the React tree. I prefer not to connect state to React that doesn't need to be there (and, for example, does not on its own trigger component updates.)

[cfoust]

What plot parameter changes cause this to happen?

Anything that affects the output points--so, basically, all plot parameter changes. Theoretically there are some kinds of mutations for which we could just modify the existing dataset, but without having the messages used to generate that dataset it's pretty tricky.

[cfoust]

Just to recap:

1. I realized that we also needed to retransmit current data to the worker when plot parameters changed, though fixing this is much simpler than the approach necessary for transmitting blocks.
2. I made it so that when a client is reset, we only send the current data it needs rather than all of the accumulated current messages.

[achim-k]

Typo: Cleint

[achim-k]

If all plot clients have a rolling data window (e.g. only the last 10 seconds are plotted), could we also filter out any current data that is not within that window?

[cfoust]

That's a bit tricky and even potentially undesirable--as a user, I would not expect that to happen. For example, users seem to rely on the fact that data stays around (even if it's not visible) for the CSV export functionality

[achim-k]

Is this guaranteed to be at max only one message per topic? Because otherwise we might drop some messages with R.map((messages) => messages.slice(-1), current) below.

[cfoust]

That's only for a single-message plot--which necessarily will ignore anything but the most recent message

[achim-k]

If all of these new current messages are already preloaded, do we still have to recompute the plot data and send it to the client? I would assume that the plot data wouldn't/shouldn't change in that case?

[cfoust]

Theoretically yes, but it's a pretty negligible amount of compute, worth fixing later only if we really think it's a problem

[jtbandes]

Just for my understanding: in the case where we do have both blocks & current data, where is the logic that chooses to use the block data instead of the current data?

[cfoust]

For generating the final dataset? That happens here

[jtbandes]

What changed in this section? I am getting a bit lost in the details and might be missing the big picture. Is it that we now use applyBlockUpdate instead of splatting the pending data into blocks?

[cfoust]

Yup, exactly. It was weird and complicated to have two different code paths for ingesting block data, one that used messages and one that used BlockUpdates.

[jtbandes]

I think I've understood what this is doing (though maybe I am missing something). I find there seems to be a lot of dancing around the difficulty of trying to partition each item in a list and then extract or sort of "transpose" the results to end up with two lists.

I think it would be shorter, clearer, and more efficient if structured more like below – unless there is some subtlety that I have lost – but if there is, I think it is too subtle because I can't see it after several minutes of reading the code.

const newPending: BlockUpdate[] = [];
const updatesToApply: BlockUpdate[] = [];
for (const update of pending) {
  for (const { id: clientId } of update.updates) {
    if (clientIds.includes(clientId)) {
      updatesToApply.push(update);
    } else {
      newPending.push(update);
    }
  }
}

(I'm also not quite sure I understand why we have two levels of nested collections – pending updates where each update has its own array of updates?)

[cfoust]

While this section could use some improvement--no denying that--your code snippet copies the BlockUpdate (including all of its updates) on every iteration. The subtlety is that ClientUpdates refer to data inside of the messages field on the BlockUpdate; you cannot just combine BlockUpdates as it seems you're suggesting in the parenthetical at the end of your comment.

I can think of ways to clean this up, however! It's a good call-out that this is hard to understand.

[jtbandes]

What do you mean by copy? I don’t see anything in my snippet that would be creating new objects, just organizing them into new arrays right?

[cfoust]

“copy” here meaning “duplicate”, not in terms of memory layout. You are appending the entire BlockUpdate to the list for every ClientUpdate

[cfoust]

Which, though it’s clearly just copying the reference, is a little counterintuitive since then you have to reaggregate them when you apply all the pending updates for a client that joins later

[jtbandes]

I see, I think I get your point now. Would this be more accurate?

const newPending: BlockUpdate[] = [];
const updatesToApply: BlockUpdate[] = [];
for (const blockUpdate of pending) {
  const [used, unused] = R.partition(
    ({ id: clientId }) => clientIds.includes(clientId),
    blockUpdate.updates,
  );
  if (used.length > 0) {
    updatesToApply.push({ ...blockUpdate, updates: used });
  }
  if (unused.length > 0) {
    newPending.push({ ...blockUpdate, updates: unused });
  }
}

[jtbandes]

It sounds like this means we don't necessarily want to apply the updates in order – why is that? I could imagine that if there is a list of updates with a reset=true in the middle, then it should theoretically reset things that were aggregated before it but not after it.

[cfoust]

Maybe I could state this more clearly, but a single BlockUpdate contains a single "step" forward in the block state machine. In other words, this code cannot result in out-of-order application. The scenario it's avoiding is where you apply ClientUpdates from the updates field one at a time. If one update contains shouldReset=true, all of them will for a given client, but if you apply them one by one, you might reset the client's data entirely (with shouldReset) several times.

ex, this goes from:

shouldReset update shouldReset update

to

shouldReset update update

[jtbandes]

Random drive-by thought as I'm reading this – if we are usually accessing updates by specific client ids, would it make sense to store them grouped by id instead of in a single array?

[cfoust]

While this is totally subjective and I'm not going to die on this hill, I have a mild preference in favor of lists over associative arrays, mostly because it makes mapping and filtering more ergonomic. In this case, however, it may make sense given that we just groupBy in applyBlockUpdate anyway, but I will think about it

[jtbandes]

I'm mildly concerned that a proliferation of reduce with object spreads will lead to a lot more CPU work being done than it would be with a loop more like const clientMessages = {}; for (const clientUpdate of updates) { clientMessages[clientUpdate.update.topic] = ... }. In less-hot code paths or when N is small I'm sure it's not a big deal but I am gathering that this will be happening rapidly many times in a row for each topic during preloading?

[cfoust]

In the worst case this code block runs N*M times (for each client) while preloading data, where N is the number of blocks, M is the number of topics, so usually this is pretty small. In relative terms the compute cost of this is dwarfed by dataset generation and downsampling, where runtime complexity is obviously a function of the number of messages.

[jtbandes]

Just for my understanding: in the case where we do have both blocks & current data, where is the logic that chooses to use the block data instead of the current data?

[achim-k]

Should we also have an upper limit for memory size? Depending on the message size, 50k messages might be way more than the allowed JS heap size, so keeping on to the full message objects should be avoided.

[defunctzombie]

Possibly - though I think we use typedarrays for storing this data so it won't be the same kind of impact on the heap?

[achim-k]

Ah sorry, you are right. I was referring to the wrong code section, I actually meant

studio/packages/studio-base/src/panels/Plot/useDatasets.ts

Line 137 in b31eca4

datasetsState = updateCurrent(messages, datasetsState);

where we concatenate new current message events to existing ones, and there seems to be no culling in place like it is here. I'm pretty sure that this is one of the main reasons for OOM crashes.

[achim-k]

-> #7298