我经常使用带有阻塞直到完成的方法的 python 对象，并希望将这些方法转换为非阻塞版本。我发现自己经常执行以下模式:
- 定义对象
- 定义一个函数来创建对象的实例，并解析命令以调用对象的方法
- 定义一个“父”对象，它创建一个运行第 2 步中定义的函数的子进程，并复制原始对象的方法。
这完成了工作，但涉及大量乏味的重复代码，而且对我来说似乎不是很 Pythonic。 是否有更好的标准方法来执行此操作？
一个高度简化的例子来说明我一直在使用的模式:
import ctypes
import Queue
import multiprocessing as mp
class Hardware:
def __init__(
self,
other_init_args):
self.dll = ctypes.cll.LoadLibrary('hardware.dll')
self.dll.Initialize(other_init_args)
def blocking_command(self, arg_1, arg_2, arg_3):
"""
This command takes a long time to execute, and blocks while it
executes. However, while it's executing, we have to coordinate
other pieces of hardware too, so blocking is bad.
"""
self.dll.Takes_A_Long_Time(arg_1, arg_2, arg_3)
def change_settings(self, arg_1, arg_2):
"""
Realistically, there's tons of other functions in the DLL we
want to expose as methods. For this example, just one.
"""
self.dll.Change_Settings(arg_1, arg_2)
def close(self):
self.dll.Quit()
def hardware_child_process(
commands,
other_init_args):
hw = Hardware(other_init_args)
while True:
cmd, args = commands.recv()
if cmd == 'start':
hw.blocking_command(**args)
elif cmd == 'change_settings':
hw.change_settings(**args)
elif cmd == 'quit':
break
hw.close()
class Nonblocking_Hardware:
"""
This class (hopefully) duplicates the functionality of the
Hardware class, except now Hardware.blocking_command() doesn't
block other execution.
"""
def __init__(
self,
other_init_args):
self.commands, self.child_commands = mp.Pipe()
self.child = mp.Process(
target=hardware_child_process,
args=(self.child_commands,
other_init_args))
self.child.start()
def blocking_command(self, arg_1, arg_2, arg_3):
"""
Doesn't block any more!
"""
self.commands.send(
('start',
{'arg_1': arg_1,
'arg_2': arg_2,
'arg_3': arg_3}))
def change_settings(self, arg_1, arg_2):
self.commands.send(
('change_settings',
{'arg_1': arg_1,
'arg_2': arg_2}))
def close(self):
self.commands.send(('quit', {}))
self.child.join()
return None
背景故事:
我使用 Python 来控制硬件，通常是通过我使用 ctypes 调用的闭源 DLL。通常，我最终想要从 DLL 调用函数，该函数会阻塞直到执行完成，但我不希望我的控制代码阻塞。例如，我可能会使用模拟输出卡将相机与照明同步。必须在模拟输出卡向相机发送触发脉冲之前调用相机 DLL 的“snap”函数，但“snap”命令会阻止我激活模拟输出卡。
最佳答案
我通过使用元类在对象上创建阻塞函数的非阻塞版本来完成类似的事情。它允许您通过执行以下操作来创建类的非阻塞版本:
class NB_Hardware(object):
__metaclass__ = NonBlockBuilder
delegate = Hardware
nb_funcs = ['blocking_command']
我采用了我最初的实现，它针对 Python 3 并使用了 concurrent.futures.ThreadPoolExecutor(我包装了阻塞 I/O 调用以使其在 中成为非阻塞asyncio context*)，并调整它们以使用 Python 2 和 concurrent.futures.ProcessPoolExecutor。下面是元类及其辅助类的实现:
from multiprocessing import cpu_count
from concurrent.futures import ProcessPoolExecutor
def runner(self, cb, *args, **kwargs):
return getattr(self, cb)(*args, **kwargs)
class _ExecutorMixin():
""" A Mixin that provides asynchronous functionality.
This mixin provides methods that allow a class to run
blocking methods in a ProcessPoolExecutor.
It also provides methods that attempt to keep the object
picklable despite having a non-picklable ProcessPoolExecutor
as part of its state.
"""
pool_workers = cpu_count()
def run_in_executor(self, callback, *args, **kwargs):
""" Runs a function in an Executor.
Returns a concurrent.Futures.Future
"""
if not hasattr(self, '_executor'):
self._executor = self._get_executor()
return self._executor.submit(runner, self, callback, *args, **kwargs)
def _get_executor(self):
return ProcessPoolExecutor(max_workers=self.pool_workers)
def __getattr__(self, attr):
if (self._obj and hasattr(self._obj, attr) and
not attr.startswith("__")):
return getattr(self._obj, attr)
raise AttributeError(attr)
def __getstate__(self):
self_dict = self.__dict__
self_dict['_executor'] = None
return self_dict
def __setstate__(self, state):
self.__dict__.update(state)
self._executor = self._get_executor()
class NonBlockBuilder(type):
""" Metaclass for adding non-blocking versions of methods to a class.
Expects to find the following class attributes:
nb_funcs - A list containing methods that need non-blocking wrappers
delegate - The class to wrap (add non-blocking methods to)
pool_workers - (optional) how many workers to put in the internal pool.
The metaclass inserts a mixin (_ExecutorMixin) into the inheritence
hierarchy of cls. This mixin provides methods that allow
the non-blocking wrappers to do their work.
"""
def __new__(cls, clsname, bases, dct, **kwargs):
nbfunc_list = dct.get('nb_funcs', [])
existing_nbfuncs = set()
def find_existing_nbfuncs(d):
for attr in d:
if attr.startswith("nb_"):
existing_nbfuncs.add(attr)
# Determine if any bases include the nb_funcs attribute, or
# if either this class or a base class provides an actual
# implementation for a non-blocking method.
find_existing_nbfuncs(dct)
for b in bases:
b_dct = b.__dict__
nbfunc_list.extend(b_dct.get('nb_funcs', []))
find_existing_nbfuncs(b_dct)
# Add _ExecutorMixin to bases.
if _ExecutorMixin not in bases:
bases += (_ExecutorMixin,)
# Add non-blocking funcs to dct, but only if a definition
# is not already provided by dct or one of our bases.
for func in nbfunc_list:
nb_name = 'nb_{}'.format(func)
if nb_name not in existing_nbfuncs:
dct[nb_name] = cls.nbfunc_maker(func)
return super(NonBlockBuilder, cls).__new__(cls, clsname, bases, dct)
def __init__(cls, name, bases, dct):
""" Properly initialize a non-blocking wrapper.
Sets pool_workers and delegate on the class, and also
adds an __init__ method to it that instantiates the
delegate with the proper context.
"""
super(NonBlockBuilder, cls).__init__(name, bases, dct)
pool_workers = dct.get('pool_workers')
delegate = dct.get('delegate')
old_init = dct.get('__init__')
# Search bases for values we care about, if we didn't
# find them on the child class.
for b in bases:
if b is object: # Skip object
continue
b_dct = b.__dict__
if not pool_workers:
pool_workers = b_dct.get('pool_workers')
if not delegate:
delegate = b_dct.get('delegate')
if not old_init:
old_init = b_dct.get('__init__')
cls.delegate = delegate
# If we found a value for pool_workers, set it. If not,
# ExecutorMixin sets a default that will be used.
if pool_workers:
cls.pool_workers = pool_workers
# Here's the __init__ we want every wrapper class to use.
# It just instantiates the delegate object.
def init_func(self, *args, **kwargs):
# Be sure to call the original __init__, if there
# was one.
if old_init:
old_init(self, *args, **kwargs)
if self.delegate:
self._obj = self.delegate(*args, **kwargs)
cls.__init__ = init_func
@staticmethod
def nbfunc_maker(func):
def nb_func(self, *args, **kwargs):
return self.run_in_executor(func, *args, **kwargs)
return nb_func
用法:
from nb_helper import NonBlockBuilder
import time
class Hardware:
def __init__(self, other_init_args):
self.other = other_init_args
def blocking_command(self, arg_1, arg_2, arg_3):
print("start blocking")
time.sleep(5)
return "blocking"
def normal_command(self):
return "normal"
class NBHardware(object):
__metaclass__ = NonBlockBuilder
delegate = Hardware
nb_funcs = ['blocking_command']
if __name__ == "__main__":
h = NBHardware("abc")
print "doing blocking call"
print h.blocking_command(1,2,3)
print "done"
print "doing non-block call"
x = h.nb_blocking_command(1,2,3) # This is non-blocking and returns concurrent.future.Future
print h.normal_command() # You can still use the normal functions, too.
print x.result() # Waits for the result from the Future
输出:
doing blocking call
start blocking
< 5 second delay >
blocking
done
doing non-block call
start blocking
normal
< 5 second delay >
blocking
对您来说，一件棘手的事情是确保 Hardware 是可挑选的。你可以通过让 __getstate__ 删除 dll 对象，然后在 __setstate__ 中重新创建它，类似于 _ExecutorMixin 确实如此。
您还需要 Python 2.x backport of concurrent.futures .
请注意，元类中有很多复杂性，因此它们可以与继承一起正常工作，并支持诸如提供 __init__ 和 nb_* 的自定义实现之类的东西方法。例如，支持这样的东西:
class AioBaseLock(object):
__metaclass__ = NonBlockBuilder
pool_workers = 1
coroutines = ['acquire', 'release']
def __init__(self, *args, **kwargs):
self._threaded_acquire = False
def _after_fork(obj):
obj._threaded_acquire = False
register_after_fork(self, _after_fork)
def coro_acquire(self, *args, **kwargs):
def lock_acquired(fut):
if fut.result():
self._threaded_acquire = True
out = self.run_in_executor(self._obj.acquire, *args, **kwargs)
out.add_done_callback(lock_acquired)
return out
class AioLock(AioBaseLock):
delegate = Lock
class AioRLock(AioBaseLock):
delegate = RLock
如果你不需要那种灵 active ，你可以大大简化实现:
class NonBlockBuilder(type):
""" Metaclass for adding non-blocking versions of methods to a class.
Expects to find the following class attributes:
nb_funcs - A list containing methods that need non-blocking wrappers
delegate - The class to wrap (add non-blocking methods to)
pool_workers - (optional) how many workers to put in the internal pool.
The metaclass inserts a mixin (_ExecutorMixin) into the inheritence
hierarchy of cls. This mixin provides methods that allow
the non-blocking wrappers to do their work.
"""
def __new__(cls, clsname, bases, dct, **kwargs):
nbfunc_list = dct.get('nb_funcs', [])
# Add _ExecutorMixin to bases.
if _ExecutorMixin not in bases:
bases += (_ExecutorMixin,)
# Add non-blocking funcs to dct, but only if a definition
# is not already provided by dct or one of our bases.
for func in nbfunc_list:
nb_name = 'nb_{}'.format(func)
dct[nb_name] = cls.nbfunc_maker(func)
return super(NonBlockBuilder, cls).__new__(cls, clsname, bases, dct)
def __init__(cls, name, bases, dct):
""" Properly initialize a non-blocking wrapper.
Sets pool_workers and delegate on the class, and also
adds an __init__ method to it that instantiates the
delegate with the proper context.
"""
super(NonBlockBuilder, cls).__init__(name, bases, dct)
pool_workers = dct.get('pool_workers')
cls.delegate = dct['delegate']
# If we found a value for pool_workers, set it. If not,
# ExecutorMixin sets a default that will be used.
if pool_workers:
cls.pool_workers = pool_workers
# Here's the __init__ we want every wrapper class to use.
# It just instantiates the delegate object.
def init_func(self, *args, **kwargs):
self._obj = self.delegate(*args, **kwargs)
cls.__init__ = init_func
@staticmethod
def nbfunc_maker(func):
def nb_func(self, *args, **kwargs):
return self.run_in_executor(func, *args, **kwargs)
return nb_func
* 原码为here , 供引用。
关于python - 制作对象的非阻塞版本的 Pythonic 方法是什么？，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/26020340/