Remove asyncio queue from control_loop #315
Conversation
🦋 Changeset detectedLatest commit: 9076a24 The changes in this PR will be included in the next version bump. Not sure what this means? Click here to learn what changesets are. Click here if you're a maintainer who wants to add another changeset to this PR |
adrianlyjak
force-pushed
the
adrian/ticky
branch
6 times, most recently
from
4b797ce to
8c4d60a
Compare
adrianlyjak
force-pushed
the
adrian/ticky
branch
from
8c4d60a to
30278d7
Compare
Pull Request Test Coverage Report for Build 21552592448Details
💛 - Coveralls |
adrianlyjak
force-pushed
the
adrian/ticky
branch
3 times, most recently
from
cb8cc07 to
e95d13d
Compare
adrianlyjak
force-pushed
the
adrian/ticky
branch
from
d89ef9f to
c9b2ccb
Compare
![devin-ai-integration[bot]](https://avatars.githubusercontent.com/in/811515?s=60&v=4){kind=small}
| return await self.queue.get() | ||
| def schedule_tick(self, tick: WorkflowTick, at_time: float) -> None: | ||
| """Schedule a tick to be processed at a specific time.""" | ||
| heapq.heappush(self.scheduled_wakeups, (at_time, tick)) |
There was a problem hiding this comment.
🔴 Heap comparison TypeError when two ticks are scheduled at the same time
The scheduled_wakeups heap stores tuples of (float, WorkflowTick). When two ticks are scheduled at exactly the same timestamp, Python's heapq will try to compare the WorkflowTick objects as a tiebreaker. However, WorkflowTick dataclasses (defined in workflows/runtime/types/ticks.py:27-70) don't implement __lt__, causing a TypeError: '<' not supported between instances.
Click to expand
How it triggers
At control_loop.py:144, heapq.heappush(self.scheduled_wakeups, (at_time, tick)) is called. If two ticks have the same at_time value, the heap will attempt to compare the ticks:
# control_loop.py:144
heapq.heappush(self.scheduled_wakeups, (at_time, tick))This scenario can occur when:
- Multiple delayed events are scheduled at the same time (e.g., retry delays)
- Timeout and another tick happen to have identical timestamps
Reproduction
import heapq
from dataclasses import dataclass
@dataclass(frozen=True)
class TickTimeout:
timeout: float
heap = []
heapq.heappush(heap, (1.0, TickTimeout(timeout=1.0)))
heapq.heappush(heap, (1.0, TickTimeout(timeout=2.0))) # Same timestamp
# TypeError: '<' not supported between instances of 'TickTimeout' and 'TickTimeout'Impact
Workflow crashes with TypeError when two events happen to be scheduled at the exact same time.
Recommendation: Add a sequence number as a tiebreaker in the tuple: (at_time, sequence_counter, tick). Increment the counter on each push. Alternatively, wrap ticks in a class that implements __lt__ based on insertion order.
Was this helpful? React with 👍 or 👎 to provide feedback.
| class _DelayedTick: | ||
| """Sentinel class representing a tick that should be scheduled after a delay.""" | ||
|
|
||
| def __init__(self, tick: WorkflowTick, delay: float) -> None: | ||
| self.tick = tick | ||
| self.delay = delay |
There was a problem hiding this comment.
🚩 Dead code: _DelayedTick class is defined but never used
The _DelayedTick class at control_loop.py:77-82 is defined but never instantiated or referenced anywhere in the codebase. This appears to be leftover code from the refactoring. The control loop now uses schedule_tick() with a heap-based approach instead. This is harmless but could be cleaned up.
Was this helpful? React with 👍 or 👎 to provide feedback.
Splitting off some changes from DBOS work
Control Loop Changes
Misc Changs