我最终使用gipc
了启动流程。取消机制依赖于这样一个事实,即当gipc
进程终止时,管道关闭。整个 API 很复杂,这就是我最终得到的结果:
class ZerorpcService():
def __init__(self):
self.participant_id = None
self.extraction_methods = []
# maps pid to (process, pipe writer)
self.processes = {}
self. = lock.Semaphore()
def _launch_process(self, function, kwargs):
"""
Launches a new process
"""
try:
# add required arguments
pid = kwargs["pid"]
# start independent gipc process, communicated via pipe
started = False
with gipc.pipe() as (r, w):
with self.mutex:
if pid in self.processes:
return_value = {'status': 1, 'error': 'pid already exists', "report": True}
return
proc = gipc.start_process(self._process_wrapper, args=(function, kwargs, w))
self.processes[pid] = proc
started = True
# wait for process to send something over pipe
return_value = r.get()
except EOFError as eof:
# happens when we terminate a process because the pipe closes
return_value = {'status': 1, 'error': "pid {} terminated".format(pid), "report": False}
except Exception as error:
logging.exception(error)
return_value = {'status': 1, 'error': str(error), 'traceback': traceback.format_exc(), "report": True}
finally:
# deletes the pid from the map
with self.mutex:
if started:
del self.processes[pid]
return return_value
@staticmethod
def _process_wrapper(function, kwargs, pipe):
"""
Executes f with kwargs and formats the result into a dict.
Wraps it in error handling.
Routes the return value through the pipe provided.
"""
return_val = {'status': 0}
try:
raw_val = function(**kwargs)
if raw_val is not None:
return_val = raw_val
except Exception as error:
logging.exception(error)
return_val = {'status': 1, 'error': str(error), 'traceback': traceback.format_exc(), "report": True}
finally:
pipe.put(return_val)
def cancel_process(self, pid):
if pid in self.processes:
with self.mutex:
process = self.processes[pid]
if process.is_alive():
process.terminate()
return {'status': 0}
else:
return {'status': 1, 'error': 'pid {} not found'.format(pid), "traceback": traceback.format_exc(),
"report": True}