Ensure scheduler events do not hold on to TaskState objects

[fjetter]

ws.processing: dict[TaskState, float]

These event logs hold on to TaskState objects which is not great because they can have a bit of a memory footprint due to the run_spec.
I initially thought this would also pose a problem about GC but we're clearing all references if a task is forgotten so this should not hold indirect references to the entire graph.

I know the test is a bit weird but it does capture this problem ¯_(ツ)_/¯

[mrocklin]

🤔 should we track instances of TaskState, perhaps with TaskState._instances = WeakSet() and verify the following in a validation function:

len(TaskState._instances) <= 2*sum(len(s.tasks) for s in Scheduler._instances)

This might be a good check for leaks more broadly.

[github-actions]

Unit Test Results

       16 files  ±0         16 suites  ±0   7h 8m 31s ⏱️ - 42m 8s
  2 741 tests  - 1    2 657 ✔️  - 3       80 💤  - 1  4 ❌ +3 
21 829 runs  +8  20 777 ✔️ +2  1 048 💤 +3  4 ❌ +3 

For more details on these failures, see this check.

Results for commit afac239. ± Comparison against base commit b6a62f8.

♻️ This comment has been updated with latest results.

[mrocklin]

@fjetter should I merge this in or are you planning to do more work here?

[fjetter]

I think adding the Task instance tracking would be a nice things to have. I'll have a look and add it if nothing breaks immediately

[fjetter]

should we track instances of TaskState, perhaps with TaskState._instances = WeakSet()

I tried but will drop this for now. I think this is worth spending some time on but it's a bit messy right now #6246

len(TaskState._instances) <= 2*sum(len(s.tasks) for s in Scheduler._instances)

FWIW I think this should len(TaskState._instances) == sum(len(s.tasks) for s in Scheduler._instances)
since every TaskState per scheduler should be unique and instances should not be shared across schedulers.