Repository navigation
Conversation
vdusek
added
the
t-tooling
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## master #1154 +/- ##
==========================================
+ Coverage 93.42% 93.59% +0.17%
==========================================
Files 54 54
Lines 3771 3858 +87
==========================================
+ Hits 3523 3611 +88
+ Misses 248 247 -1
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
5bbadc2 to
033624c
Compare
7322d8e to
7c3d462
Compare
7c3d462 to
6f9300e
Compare
6f9300e to
c2f5a87
Compare
apify-service-account
added
the
tested
Pijukatel
left a comment
Pijukatel
left a comment
There was a problem hiding this comment.
I am struggling a lot with the variable naming in this PR. Reading this with fresh eyes the variable names gave me a lot of false leads instead of building a helpful mental model.
The usage of word slot here had a very confusing effect on me. One time it was for an asyncio.Condition and the second time for a very custom context manager that uses the first one inside...
On top of it, slot has its own established meaning in Python, but here it is not related in any manner.
| """Client the run under each name was last started or looked up with. Lost on a migration.""" | ||
| self._max_concurrent_runs: int | None = None | ||
| """How many recorded runs may be active at once, or `None` for no limit.""" | ||
| self._slots = asyncio.Condition() |
There was a problem hiding this comment.
This sounds more a like an iterable, not a condition name. Can you please use some condition like name.
Maybe something like _slots_available ?
There was a problem hiding this comment.
renamed to _capacity_available
| """ | ||
| return self._child_run_registry.run_clients(self.apify_client) | ||
|
|
||
| def set_child_run_limits(self, *, max_concurrent_runs: int | None) -> None: |
There was a problem hiding this comment.
The named in the description is critical. I would probably put it to the method name, because without reading the description you will get a wrong idea from just the method name. It is pretty uggly and explicit, but better than giving a wrong idea:
set_named_child_run_limits
There was a problem hiding this comment.
agreed, renamed to set_named_child_run_limits
| """How many recorded runs may be active at once, or `None` for no limit.""" | ||
| self._slots = asyncio.Condition() | ||
| """Serializes counting the active runs, and wakes the starts waiting for a slot when one may have freed up.""" | ||
| self._starting: set[str] = set() |
There was a problem hiding this comment.
This sounds like a state, not a set. Can we have a noun, so that we know what is starting?
Maybe something like _starting_names?
There was a problem hiding this comment.
ok, renamed to _starting_names
| self._observe(name, run) | ||
| if record.status != run.status: | ||
| await self._save(name, record.model_copy(update={'status': run.status})) | ||
| async with self._slots: |
There was a problem hiding this comment.
To me the naming is extremely confusing. We have these similarly named context managers. async with self._slots, async with self._slot, but they are quite something different.
When having a singluar item _slot, than I expect that _slots will be just some structure holding multiple of _slot, but here it is just completelly different object.
There was a problem hiding this comment.
Ok - the two are now _capacity_available (the condition) and _reserve_capacity() (the context manager), so they don't read as singular and plural of the same thing anymore.
| """Serializes counting the active runs, and wakes the starts waiting for a slot when one may have freed up.""" | ||
| self._starting: set[str] = set() | ||
| """Names holding a slot for a start or resurrection in flight, not counted from the records yet.""" | ||
| self._observed: dict[str, tuple[str, float]] = {} |
There was a problem hiding this comment.
observed what? The name for this will be even more difficult, but observed tells too little...
Maybe timestamped_run_statuses?
There was a problem hiding this comment.
that's what docstrings are for 🙂
self._observed: dict[str, tuple[str, float]] = {}
"""Last observed status of the run recorded under each name, with the event loop time it was observed at."""
but ok, _timestamped_run_statuses it is
Edit: in the end it is _run_status_cache and a new NamedTuple type, I believe this is even better
… and applies a new limit
Adds
Actor.set_named_child_run_limits(max_concurrent_runs=...). While that many named child runs areREADY,RUNNING,ABORTINGorTIMING-OUT, a namedActor.start,Actor.call,Actor.start_taskorActor.call_taskthat would start or resurrect a run waits for one of them to finish. Reattaching to a recorded run never waits, and runs without arun_namearen't counted.The count comes from the child run registry, so children started before a migration count too. A child's status is fetched again before counting once it's more than 10 seconds old, so a finished run nobody waited on stops counting.
Actor.callandActor.call_taskstop counting their run as soon as it finishes. A start counts toward the limit until its run is recorded, which keeps concurrent starts under different names within the limit.The limit lives in memory, so the Actor sets it again after a migration. Once the parent receives
ABORTING, a start still waiting under the limit raises aRuntimeErrorwithout starting a run. Waiting further would block the abort of the other children.Closes: #1138
✍️ Drafted by Claude Code