AOSSIE-Org / AOSSIE-Org/Devr.AI

ENHANCEMENT:Queue Worker Inefficient Polling

Ouverte
#251 3 commentaires 0 réactions 0 personnes assignées Voir sur GitHub
Langage dominant
Python
Étoiles
102
Forks
137
Métriques de merge des PR
Aucune PR mergée en 30 j

Description

### 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.

Guide de contribution

Aucun guide de contribution indexé pour ce dépôt

Évaluation

Cette issue n'a pas encore été évaluée.

Recevez les nouvelles issues par e-mail

Un résumé court des issues GitHub adaptées aux débutants.