asyncio 中 loop.run_forever() 方法导致 100% CPU usage

def main(): log_listener = setup_logging(log_filename) e = asyncio.Event() cOnsumer= Consumer(e) loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) signals = (signal.SIGHUP, signal.SIGTERM, signal.SIGINT) for sig in signals: loop.add_signal_handler( sig, lambda s=sig: asyncio.create_task(shutdown(s, loop, consumer)) ) tasks = consumer.run() try: for name, task in tasks.items(): loop.create_task(task, name=name) loop.run_forever() finally: loop.close() log_listener.stop()

consumer.run() 方法会返回一个 Dict[str, Coroutine] 类型的字典。最初以为是自己的 coroutine 实现有问题导致 high CPU 。然后将 tasks 中的 coroutine 一个个移除，最后 tasks 返回空的情况下也是 100% CPU 。
用 cProfile 看了下：

 Ordered by: cumulative time List reduced from 3038 to 10 due to restriction <10> ncalls tottime percall cumtime percall filename:lineno(function) 423/1 0.005 0.000 44.862 44.862 {built-in method builtins.exec} 1 0.000 0.000 44.862 44.862 myscript.py:1(<module>) 1 0.000 0.000 44.219 44.219 myscript.py:54(main) 1 0.000 0.000 44.017 44.017 /home/xxx/.pyenv/versions/3.10.1/lib/python3.10/asyncio/base_events.py:582(run_forever) 3 0.000 0.000 44.016 14.672 /home/xxx/.pyenv/versions/3.10.1/lib/python3.10/asyncio/base_events.py:1806(_run_once) 3 0.000 0.000 43.983 14.661 /home/xxx/.pyenv/versions/3.10.1/lib/python3.10/selectors.py:452(select) 3 43.983 14.661 43.983 14.661 {method 'poll' of 'select.epoll' objects} 484/4 0.006 0.000 0.642 0.161 <frozen importlib._bootstrap>:1022(_find_and_load) 483/4 0.003 0.000 0.642 0.161 <frozen importlib._bootstrap>:987(_find_and_load_unlocked) 447/5 0.003 0.000 0.641 0.128 <frozen importlib._bootstrap>:664(_load_unlocked)

有 v 友遇到类似情况的吗？还是说我的用法有问题。

Supplement 1 Apr 7, 2022

 def _start_kafka_client(self) -> None: logging.debug(f"[_start_kafka_client] id(event)={id(self._event)}") i = 0 try: while not self._event.is_set(): msg_packs = self._kafka_client.poll( timeout_ms=1000, max_records=5000, ) if not msg_packs: continue # msgs is of type list containerd with ConsumerRecords tp: kafka.TopicPartition msgs: List[ConsumerRecord] for tp, msgs in msg_packs.items(): self._data_queue.put_nowait(msgs) i += len(msgs) if i % 1000 == 0: logging.info(f"count of msgs: {i}") except asyncio.QueueFull: logging.debug("data queue is full") time.sleep(1) except Exception as e: logging.error(f"error: {e}") finally: logging.debug("_start_kafka_client terminates")

Supplement 2 Apr 7, 2022

``` python
async def import_data(self):
logging.debug(f"id(event)={id(self._event)}")
while not self._event.is_set():
try:
if not self._template_created:
await asyncio.sleep(1)

entry: List[ConsumerRecord] = self._data_queue.get_nowait()
entries = [
each_entry
for each in entry
for each_entry in self.reconstruct_entries_list(each.value)
]

await self._es_client.bulk(
index=self.get_index_name(), operatiOns=entries
)
except asyncio.QueueEmpty:
await asyncio.sleep(8)
continue
except Exception as e:
logging.error(f"[*] error: {e}")
except asyncio.CancelledError:
logging.debug("import_data is cancelled")
break
logging.debug("import_data terminates")

def run(self) -> Dict[str, Coroutine]:
return {
"template_task": self._check_template(),
"kafka_task": asyncio.to_thread(self._start_kafka_client),
"es_task": self.import_data(),
"monitor_task": self.monitor(),
}

async def monitor(self):
try:
while not self._event.is_set():
tasks = asyncio.all_tasks()
for task in tasks:
print(f"[*] task {task.get_name()} done? {task.done()}")
await asyncio.sleep(8)
finally:
logging.debug("monitor finished")
```

8 replies 2022-04-1223:35:44 +08:00

netcan

Apr 7, 2022

tasks 为空的话，`loop.run_forever()`直接就返回了，建议上完整代码，例如 Consumer

Nitroethane

Apr 7, 2022

@netcan #1 已添加。就是从 kafka 接收数据，解析之后写到 es 里

makerbi

Apr 7, 2022

应该是 while 里没有 sleep 的原因吧
time.sleep(0.1)也好过完全没有 sleep

jenlors

Apr 7, 2022

从 kafka 读取消息的时候 block 设置为 True ，直接阻塞循环

Nitroethane

Apr 7, 2022

@makerbi #3 是读 kafka 数据的那个 while 循环吗？我以为用 poll() 方法的时候设置个 1s 的 timeout 也算 sleep 。我试试加个 sleep 有没有效果。

@jenlors #4 之前不直接阻塞的原因是不知道怎样在终止脚本运行的时候优雅地退出。今天发现在阻塞的情况下应该是可以通过捕获 CanceledError 来实现优雅退出 coroutine 。多谢回复，我试试

Richard14

Apr 10, 2022

问题太长，且缺乏最小实现，1L 代码里很多不明实现的东西，实在是不想看。而且问题给人感觉很像 AB 问题

Nitroethane

Apr 10, 2022

@Richard14 #6 我寻思也没强制你看呀,不想看就别回复呀 :)

lolizeppelin

Apr 12, 2022

不建议 sleep 0.1 ，sleep 0.001 都不适合。
一般来说都是通过监听事件 fd 来实现 sleep 的同时能及时响应