Рекомендации по архитектуре для отменяемых внутренних вызовов [zerorpc, если возможно?] - PullRequest
Новая
       30

Рекомендации по архитектуре для отменяемых внутренних вызовов [zerorpc, если возможно?]

1 голос
/ 09 июля 2019

У меня есть веб-интерфейс электронного приложения (zerorpc-node), взаимодействующий с бэкэндом Python (zerorpc-python), который должен:

a) иметь возможность отправлять запросы в бэкэнд [стандартный вызов zerorpc]

b) иметь возможность запускать несколько бэкэнд-процессов одновременно [следуя архитектуре в https://github.com/0rpc/zerorpc-node/issues/96)

c) иметь возможность отменить бэкэнд-процесс по желанию [не уверен, как это сделатьс текущей архитектурой]

Любое руководство о том, как создать архитектуру решения (c), было бы замечательно.При необходимости я готов отказаться от zerorpc, если он ограничивает, но если решение предполагает использование zerorpc, это фантастика.

1 Ответ

0 голосов
/ 18 июля 2019

Я использовал 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}
...