Drain the queue into the sinks in batches until cancelled.
()
| 308 | |
| 309 | |
| 310 | async def _writer_loop() -> None: |
| 311 | """Drain the queue into the sinks in batches until cancelled.""" |
| 312 | assert _queue is not None |
| 313 | |
| 314 | while True: |
| 315 | try: |
| 316 | batch = [await _queue.get()] |
| 317 | # One blocking get above, then take whatever else has piled up |
| 318 | # without waiting. Under light traffic this writes a single row |
| 319 | # after the interval; under load it fills whole batches. |
| 320 | deadline = asyncio.get_running_loop().time() + BATCH_INTERVAL |
| 321 | while len(batch) < BATCH_SIZE: |
| 322 | timeout = deadline - asyncio.get_running_loop().time() |
| 323 | if timeout <= 0: |
| 324 | break |
| 325 | try: |
| 326 | batch.append(await asyncio.wait_for(_queue.get(), timeout)) |
| 327 | except asyncio.TimeoutError: |
| 328 | break |
| 329 | |
| 330 | await _flush(batch) |
| 331 | |
| 332 | except asyncio.CancelledError: |
| 333 | raise |
| 334 | except Exception as e: |
| 335 | logger.error(f"Usage log writer error: {e}") |
| 336 | await asyncio.sleep(1) |
| 337 | |
| 338 | |
| 339 | # --- history ----------------------------------------------------------------- |