Repository navigation
fix(polling): bound active workers across polling iterations - #3063
IsmaelMartinez merged 1 commit into
Conversation
fd0df9e to
a9dac92
Compare
a9dac92 to
cd9bb86
Compare
cd9bb86 to
2036747
Compare
IsmaelMartinez
left a comment
There was a problem hiding this comment.
Thanks, the cross-iteration cap is the right fix: on main the ten-worker limit restarts from zero every poll and nothing is ever joined. Two inline points, both small.
| except BaseException as exc: | ||
| # Treat a PID-bearing child as potentially dispatched; never retry its task. | ||
| if process.pid is not None: | ||
| active_processes.append(process) | ||
| task_queue.popleft() | ||
| else: | ||
| process.close() | ||
| if isinstance(exc, Exception): | ||
| raise _PollingWorkerStartError("Polling worker startup failed; stopping dispatch") from exc | ||
| raise |
There was a problem hiding this comment.
When pid is None the fork never happened, so nothing was dispatched and there is no duplicate to protect against, yet a transient EAGAIN from start() still exits the service with code 1 where main logged and kept polling. Could this branch drop the batch and continue, and keep the fail-stop for the pid-bearing case?
| except BaseException as exc: | |
| # Treat a PID-bearing child as potentially dispatched; never retry its task. | |
| if process.pid is not None: | |
| active_processes.append(process) | |
| task_queue.popleft() | |
| else: | |
| process.close() | |
| if isinstance(exc, Exception): | |
| raise _PollingWorkerStartError("Polling worker startup failed; stopping dispatch") from exc | |
| raise | |
| except BaseException as exc: | |
| # Treat a PID-bearing child as potentially dispatched; never retry its task. | |
| if process.pid is not None: | |
| active_processes.append(process) | |
| task_queue.popleft() | |
| if isinstance(exc, Exception): | |
| raise _PollingWorkerStartError("Polling worker startup failed; stopping dispatch") from exc | |
| raise | |
| process.close() | |
| if not isinstance(exc, Exception): | |
| raise | |
| # No child exists, so nothing can be duplicated: drop the batch and keep polling. | |
| get_logger().error(f"Polling worker startup failed before dispatch: {exc}") | |
| break |
There was a problem hiding this comment.
Keeping the fail-stop behavior here. Process.pid is None does not reliably prove that no OS child was created, since _popen is assigned only after _Popen(...) returns successfully. Continuing polling in that uncertain state could admit more work or risk duplicate processing.
| try: | ||
| while task_queue: | ||
| _reap_finished_processes(active_processes) | ||
| if len(active_processes) >= max_allowed_parallel_tasks: |
There was a problem hiding this comment.
Waiting on capacity emits nothing and has no upper bound, since the children carry no timeout, so a poller correctly waiting looks the same as a hung one. One line on first entry would make it diagnosable.
| try: | |
| while task_queue: | |
| _reap_finished_processes(active_processes) | |
| if len(active_processes) >= max_allowed_parallel_tasks: | |
| waiting_logged = False | |
| try: | |
| while task_queue: | |
| _reap_finished_processes(active_processes) | |
| if len(active_processes) >= max_allowed_parallel_tasks: | |
| if not waiting_logged: | |
| waiting_logged = True | |
| get_logger().info( | |
| f"Polling dispatch waiting for capacity: {len(active_processes)} workers active, " | |
| f"{len(task_queue)} tasks queued") |
There was a problem hiding this comment.
Good point. Added a one-time info log when dispatch first waits for worker capacity, including active-worker and queued-task counts. It does not repeat on every 250 ms check, and the behavior is covered by the existing test.
2036747 to
93862f4
Compare
IsmaelMartinez
left a comment
There was a problem hiding this comment.
Thanks, and your reasoning on the fail-stop is better than mine: _popen is assigned only after _Popen returns, so a None pid genuinely does not prove no child was forked. Keeping the fail-stop is the right call and I withdraw that one.
The one-time capacity log is what I was after. Suite 4389 to 4402, and the revert probe is fully red: all 14 tests fail against main's github_polling.py.
Summary
Apply the existing ten-worker GitHub polling limit across polling iterations, not just within each batch. Track active child processes, reap completed workers, and wait asynchronously when earlier work already consumes the available capacity.
Behavior
Waiting work remains memory-only and notification acknowledgement is unchanged; this does not add durable delivery.
Validation
This change is independent of #2624 and the separate async HTTP fallback PR.