AOSSIE-Org / AOSSIE-Org/Devr.AI
ENHANCEMENT:Queue Worker Inefficient Polling
- 主要言語
- Python
- スター
- 102
- フォーク
- 137
- PR マージ指標
- 30日以内にマージされた PR はありません
説明
### Is there an existing issue for this?
- [x] I have searched the existing issues
### What happened?
Queue workers use inefficient polling with `asyncio.sleep(0.1)` in a tight loop, causing CPU waste and up to 100ms message processing delays.
**Location**: `backend/app/core/orchestration/queue_manager.py:99-124`
### Current Behavior
```python
async def _worker(self, worker_name: str):
queues = [...]
while self.running:
for queue in queues:
message = await queue.get(no_ack=False, fail=False)
if message:
# process message...
await asyncio.sleep(0.1) # ❌ Polls every 100ms even when empty
```
**Impact**:
- 30 polls/second (10 per worker × 3 workers) when queues are empty
- Up to 100ms delay before messages are processed
- Unnecessary CPU usage during idle periods
### Expected Behavior
Workers should block and wait for messages instead of polling:
- Near-instant message processing (eliminate 100ms delay)
- ~90% reduction in CPU usage during idle
- Better resource efficiency
### Proposed Solution
Use `asyncio.wait()` to block until messages arrive:
```python
async def _worker(self, worker_name: str):
queues = [
await self.channel.declare_queue(self.queues[priority], durable=True)
for priority in [QueuePriority.HIGH, QueuePriority.MEDIUM, QueuePriority.LOW]
]
while self.running:
try:
done, pending = await asyncio.wait(
[queue.get(no_ack=False) for queue in queues],
return_when=asyncio.FIRST_COMPLETED,
timeout=1.0
)
if not self.running:
break
for task in done:
message = await task
if message:
item = json.loads(message.body.decode())
await self._process_item(item, worker_name)
await message.ack()
for task in pending:
task.cancel()
except asyncio.CancelledError:
return
except Exception as e:
logger.error(f"Worker {worker_name} error: {e}")
await asyncio.sleep(1)
```
### Testing Checklist
- [x] Messages processed immediately when enqueued
- [x] CPU usage near zero during idle periods
- [x] Priority handling (HIGH → MEDIUM → LOW) maintained
- [x] No message duplication with multiple workers
- [x] Graceful shutdown works correctly
### Performance Targets
- CPU reduction: ~90% during idle
- Latency: <10ms (from current ~100ms)
- Throughput: Maintain or improve
### Additional Context
This is part of a broader performance optimization effort. The current implementation works but is inefficient.
コントリビューションガイド
このリポジトリのコントリビューションガイドは索引されていません
評価
この issue はまだ評価されていません。