使用线程并发调用Salt客户端API时出现无事件循环错误的解决方案咨询

我在使用线程执行salt.client.get_local_client()并并发调用cmd()函数时，遇到了线程中缺少事件循环的错误。不确定这个问题是出在我的实现方式上，还是Salt本身的设计限制。

错误日志 #

python test.py 
Exception in thread Thread-2:
Traceback (most recent call last):
Exception in thread Thread-1:
  File "/home/test/.pyenv/versions/test/lib/python3.9/site-packages/salt/client/__init__.py", line 387, in run_job
Traceback (most recent call last):
  File "/home/test/.pyenv/versions/test/lib/python3.9/site-packages/salt/client/__init__.py", line 387, in run_job
    pub_data = self.pub(
  File "/home/test/.pyenv/versions/test/lib/python3.9/site-packages/salt/client/__init__.py", line 1898, in pub
    pub_data = self.pub(
  File "/home/test/.pyenv/versions/test/lib/python3.9/site-packages/salt/client/__init__.py", line 1898, in pub
    with salt.channel.client.ReqChannel.factory(
  File "/home/test/.pyenv/versions/test/lib/python3.9/site-packages/salt/channel/client.py", line 56, in factory
    with salt.channel.client.ReqChannel.factory(
    return SyncWrapper(
  File "/home/test/.pyenv/versions/test/lib/python3.9/site-packages/salt/utils/asynchronous.py", line 76, in __init__
    return SyncWrapper(
  File "/home/test/.pyenv/versions/test/lib/python3.9/site-packages/salt/utils/asynchronous.py", line 76, in __init__
    self.obj = cls(*args, **kwargs)
  File "/home/test/.pyenv/versions/test/lib/python3.9/site-packages/salt/channel/client.py", line 138, in factory
    self.obj = cls(*args, **kwargs)
  File "/home/test/.pyenv/versions/test/lib/python3.9/site-packages/salt/channel/client.py", line 138, in factory
    transport = salt.transport.request_client(opts, io_loop=io_loop)
  File "/home/test/.pyenv/versions/test/lib/python3.9/site-packages/salt/transport/base.py", line 59, in request_client
    transport = salt.transport.request_client(opts, io_loop=io_loop)
  File "/home/test/.pyenv/versions/test/lib/python3.9/site-packages/salt/transport/base.py", line 59, in request_client
    return salt.transport.zeromq.RequestClient(opts, io_loop=io_loop)
  File "/home/test/.pyenv/versions/test/lib/python3.9/site-packages/salt/transport/zeromq.py", line 1084, in __init__
    return salt.transport.zeromq.RequestClient(opts, io_loop=io_loop)
  File "/home/test/.pyenv/versions/test/lib/python3.9/site-packages/salt/transport/zeromq.py", line 1084, in __init__
    self.sending = asyncio.Lock()
  File "/home/test/.pyenv/versions/3.9.18/lib/python3.9/asyncio/locks.py", line 81, in __init__
    self.sending = asyncio.Lock()
  File "/home/test/.pyenv/versions/3.9.18/lib/python3.9/asyncio/locks.py", line 81, in __init__
    self._loop = events.get_event_loop()
  File "/home/test/.pyenv/versions/3.9.18/lib/python3.9/asyncio/events.py", line 642, in get_event_loop
    self._loop = events.get_event_loop()
  File "/home/test/.pyenv/versions/3.9.18/lib/python3.9/asyncio/events.py", line 642, in get_event_loop
    raise RuntimeError('There is no current event loop in thread %r.'
RuntimeError: There is no current event loop in thread 'Thread-2'.

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/home/test/.pyenv/versions/3.9.18/lib/python3.9/threading.py", line 980, in _bootstrap_inner
    raise RuntimeError('There is no current event loop in thread %r.'
RuntimeError: There is no current event loop in thread 'Thread-1'.

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/home/test/.pyenv/versions/3.9.18/lib/python3.9/threading.py", line 980, in _bootstrap_inner
    self.run()
  File "/home/test/.pyenv/versions/3.9.18/lib/python3.9/threading.py", line 917, in run
    self.run()
  File "/home/test/.pyenv/versions/3.9.18/lib/python3.9/threading.py", line 917, in run
    self._target(*self._args, **self._kwargs)
  File "/home/test/test.py", line 9, in test_sleep
    self._target(*self._args, **self._kwargs)
    cl.cmd('master-local', 'test.sleep', [2])
  File "/home/test/test.py", line 9, in test_sleep
  File "/home/test/.pyenv/versions/test/lib/python3.9/site-packages/salt/client/__init__.py", line 752, in cmd
    cl.cmd('master-local', 'test.sleep', [2])
  File "/home/test/.pyenv/versions/test/lib/python3.9/site-packages/salt/client/__init__.py", line 752, in cmd
    pub_data = self.run_job(
  File "/home/test/.pyenv/versions/test/lib/python3.9/site-packages/salt/client/__init__.py", line 409, in run_job
    pub_data = self.run_job(
  File "/home/test/.pyenv/versions/test/lib/python3.9/site-packages/salt/client/__init__.py", line 409, in run_job
    raise SaltClientError(general_exception)
salt.exceptions.SaltClientError: There is no current event loop in thread 'Thread-2'.
    raise SaltClientError(general_exception)
salt.exceptions.SaltClientError: There is no current event loop in thread 'Thread-1'.

测试代码 #

from threading import Thread

import salt.client
import time

cl = salt.client.get_local_client()

def test_sleep():
    cl.cmd('master-local', 'test.sleep', [2])

t1 = Thread(target=test_sleep)
t2 = Thread(target=test_sleep)

t1.start()
# time.sleep(1)  # <-- Changes the error
t2.start()

t1.join()
t2.join()

Salt版本信息 #

Salt Version:
          Salt: 3007.1
 
Python Version:
        Python: 3.9.18 (main, Feb  3 2024, 15:02:34)
 
Dependency Versions:
          cffi: 1.16.0
      cherrypy: 18.10.0
      dateutil: 2.8.2
     docker-py: Not Installed
         gitdb: Not Installed
     gitpython: Not Installed
        Jinja2: 3.1.3
       libgit2: Not Installed
  looseversion: 1.3.0
      M2Crypto: Not Installed
          Mako: Not Installed
       msgpack: 1.0.7
  msgpack-pure: Not Installed
  mysql-python: Not Installed
     packaging: 23.2
     pycparser: 2.21
      pycrypto: 2.6.1
  pycryptodome: 3.20.0
        pygit2: Not Installed
  python-gnupg: 0.5.3
        PyYAML: 6.0.1
         PyZMQ: 25.1.2
        relenv: 0.16.0
         smmap: Not Installed
       timelib: 0.3.0
       Tornado: 6.4
           ZMQ: 4.3.4
 
Salt Package Information:
  Package Type: Not Installed
 
System Versions:
          dist: centos 8.3.2011 
        locale: utf-8
       machine: x86_64
       release: 4.18.0-240.el8.x86_64
        system: Linux
       version: CentOS Linux 8.3.2011 

我的疑问 #

如何正确地使用线程实现Salt客户端API的并发调用？这是Salt设计上的限制，还是我的实现有问题？

问题原因分析 #

这个问题本质上是因为Salt的本地客户端（LocalClient）并不是线程安全的，而且它内部依赖了asyncio事件循环——而Python的asyncio事件循环是线程绑定的：主线程会自动创建事件循环，但子线程默认不会有事件循环，当多个子线程同时访问同一个LocalClient实例时，就会触发这个错误。

你注释掉的time.sleep(1)能暂时“缓解”问题，只是因为让两个线程的执行错开了，避免了同时访问客户端实例引发的冲突，但这并不是根本解决办法。

可行的解决方案 #

这里有几个靠谱的解决思路，你可以根据自己的场景选择：

1. 每个线程创建独立的LocalClient实例

既然共享实例会出问题，那我们可以让每个线程自己创建客户端，这样每个线程都有独立的资源和事件循环：

from threading import Thread

import salt.client
import time

def test_sleep():
    # 每个线程内单独创建客户端实例
    cl = salt.client.get_local_client()
    cl.cmd('master-local', 'test.sleep', [2])

t1 = Thread(target=test_sleep)
t2 = Thread(target=test_sleep)

t1.start()
t2.start()

t1.join()
t2.join()

这个方案简单直接，能有效避免线程安全问题，也是最推荐的做法。

2. 使用线程锁保护客户端调用

如果你必须共享同一个LocalClient实例（比如出于资源考虑），可以用线程锁来确保同一时间只有一个线程调用客户端的方法：

from threading import Thread, Lock

import salt.client
import time

cl = salt.client.get_local_client()
client_lock = Lock()

def test_sleep():
    with client_lock:
        cl.cmd('master-local', 'test.sleep', [2])

t1 = Thread(target=test_sleep)
t2 = Thread(target=test_sleep)

t1.start()
t2.start()

t1.join()
t2.join()

不过这个方案会让调用变成串行执行，失去了并发的意义，所以只适合特殊场景。

3. 使用Salt的异步客户端API

Salt本身提供了异步客户端接口，如果你本身就有异步代码的需求，可以直接用salt.client.get_async_local_client()，配合asyncio的并发机制（比如asyncio.gather）来实现并发，这样更符合Salt内部的设计逻辑：

import asyncio
import salt.client

async def test_sleep():
    cl = salt.client.get_async_local_client()
    await cl.cmd('master-local', 'test.sleep', [2])

async def main():
    await asyncio.gather(test_sleep(), test_sleep())

if __name__ == "__main__":
    asyncio.run(main())

总结 #

这个问题主要是因为你共享了非线程安全的LocalClient实例，加上asyncio事件循环的线程绑定特性导致的。最合理的解决方式是给每个线程创建独立的客户端实例，或者切换到Salt的异步客户端API。Salt的设计本身并没有限制并发调用，只是需要遵循它的线程安全规则来实现。

备注：内容来源于stack exchange，提问作者Tony