# Path: botoflow/core/async_traceback.py
# def print_exc(limit=None, file=None):
# def format_exc(limit=None, exception=None, tb_list=None):
# def extract_tb(limit=None):
# 
# Path: botoflow/workers/multiprocessing_executor.py
# class MultiprocessingExecutor(object):
#     """A base for all multiprocessing executors"""
# 
#     def __init__(self, worker):
#         self._worker = worker
# 
#     def start(self):
#         """Start the worker. This method does not block."""
#         log.debug("Starting worker %s", self)
# 
#         # use this simple queue (and .empty) to check if we should shut down
#         self._worker_shutdown = self._process_manager().Queue()
#         # to track the active processes
#         self._process_queue = self._process_manager().JoinableQueue()
#         self._running = True
# 
#     def stop(self):
#         """Stops the worker processes.
#         To wait for all the processes to terminate, call:
# 
#         .. code-block:: python
# 
#             worker.start()
#             time.sleep(360)
#             worker.stop()
#             worker.join()  # will block
# 
#         """
#         log.debug("Stopping worker %s", self)
#         if not self.is_running:
#             return False
# 
#         self._worker_shutdown.put(1)
# 
#     def join(self):
#         """Will wait till all the processes are terminated
#         """
#         try:
#             self._process_queue.join()
#         except KeyboardInterrupt:
#             six.print_("\nTerminating, please wait...", file=sys.stderr)
#             self._process_queue.join()
#         self._running = False
# 
#     @property
#     def initializer(self):
#         """If set, the initializer function will be called after the subprocess
#         is started with the worker object as the first argument.
# 
#         You can use this to, for example, set the process name suffix, to
#         distinguish between activity and workflow workers (when starting them
#         from the same process):
# 
#         .. code-block:: python
# 
#             from setproctitle import getproctitle, setproctitle
# 
#             def set_worker_title(worker):
#                 name = getproctitle()
#                 if isinstance(worker, WorkflowWorker):
#                     setproctitle(name + ' (WorkflowWorker)')
#                 elif isinstance(worker, ActivityWorker):
#                     setproctitle(name + ' (ActivityWorker)')
# 
#             worker.initializer = set_worker_title
#         """
#         try:
#             return self.__initializer
#         except AttributeError:
#             return lambda obj: None
# 
#     @initializer.setter
#     def initializer(self, func):
#         self.__initializer = func
# 
#     @property
#     def is_running(self):
#         """Returns True if the worker is running"""
#         if hasattr(self, '_running'):
#             return self._running
#         return False
# 
#     @classmethod
#     def _process_manager(cls):
#         return process_manager()


import multiprocessing
import multiprocessing.queues
import signal
import logging
import dill
from ..core import async_traceback
from .multiprocessing_executor import MultiprocessingExecutor
        .. code-block:: python

            worker.start(pollers=10)
            time.sleep(360)
            worker.stop()
            worker.join()  # will block

        """
        if pollers < 1:
            raise ValueError("pollers count must be greater than 0")

        super(MultiprocessingWorkflowExecutor, self).start()

        start_condition = self._process_manager().Condition()

        def run_decider(executor):
            executor._process_queue.get()
            # ignore any SIGINT, so it looks closer to threading
            signal.signal(signal.SIGINT, signal.SIG_IGN)

            process = multiprocessing.current_process()
            log.debug("Poller/decider %s started", process.name)

            while executor._worker_shutdown.empty():
                with start_condition:
                    start_condition.notify_all()
                try:
                    executor._worker.run_once()

                except Exception as err:

# Path: botoflow/core/async_traceback.py
# def print_exc(limit=None, file=None):
# def format_exc(limit=None, exception=None, tb_list=None):
# def extract_tb(limit=None):
# 
# Path: botoflow/workers/multiprocessing_executor.py
# class MultiprocessingExecutor(object):
#     """A base for all multiprocessing executors"""
# 
#     def __init__(self, worker):
#         self._worker = worker
# 
#     def start(self):
#         """Start the worker. This method does not block."""
#         log.debug("Starting worker %s", self)
# 
#         # use this simple queue (and .empty) to check if we should shut down
#         self._worker_shutdown = self._process_manager().Queue()
#         # to track the active processes
#         self._process_queue = self._process_manager().JoinableQueue()
#         self._running = True
# 
#     def stop(self):
#         """Stops the worker processes.
#         To wait for all the processes to terminate, call:
# 
#         .. code-block:: python
# 
#             worker.start()
#             time.sleep(360)
#             worker.stop()
#             worker.join()  # will block
# 
#         """
#         log.debug("Stopping worker %s", self)
#         if not self.is_running:
#             return False
# 
#         self._worker_shutdown.put(1)
# 
#     def join(self):
#         """Will wait till all the processes are terminated
#         """
#         try:
#             self._process_queue.join()
#         except KeyboardInterrupt:
#             six.print_("\nTerminating, please wait...", file=sys.stderr)
#             self._process_queue.join()
#         self._running = False
# 
#     @property
#     def initializer(self):
#         """If set, the initializer function will be called after the subprocess
#         is started with the worker object as the first argument.
# 
#         You can use this to, for example, set the process name suffix, to
#         distinguish between activity and workflow workers (when starting them
#         from the same process):
# 
#         .. code-block:: python
# 
#             from setproctitle import getproctitle, setproctitle
# 
#             def set_worker_title(worker):
#                 name = getproctitle()
#                 if isinstance(worker, WorkflowWorker):
#                     setproctitle(name + ' (WorkflowWorker)')
#                 elif isinstance(worker, ActivityWorker):
#                     setproctitle(name + ' (ActivityWorker)')
# 
#             worker.initializer = set_worker_title
#         """
#         try:
#             return self.__initializer
#         except AttributeError:
#             return lambda obj: None
# 
#     @initializer.setter
#     def initializer(self, func):
#         self.__initializer = func
# 
#     @property
#     def is_running(self):
#         """Returns True if the worker is running"""
#         if hasattr(self, '_running'):
#             return self._running
#         return False
# 
#     @classmethod
#     def _process_manager(cls):
#         return process_manager()


# Path: botoflow/workers/multiprocessing_workflow_executor.py
import multiprocessing
import multiprocessing.queues
import signal
import logging
import dill
from ..core import async_traceback
from .multiprocessing_executor import MultiprocessingExecutor

        .. code-block:: python

            worker.start(pollers=10)
            time.sleep(360)
            worker.stop()
            worker.join()  # will block

        """
        if pollers < 1:
            raise ValueError("pollers count must be greater than 0")

        super(MultiprocessingWorkflowExecutor, self).start()

        start_condition = self._process_manager().Condition()

        def run_decider(executor):
            executor._process_queue.get()
            # ignore any SIGINT, so it looks closer to threading
            signal.signal(signal.SIGINT, signal.SIG_IGN)

            process = multiprocessing.current_process()
            log.debug("Poller/decider %s started", process.name)

            while executor._worker_shutdown.empty():
                with start_condition:
                    start_condition.notify_all()
                try:
                    executor._worker.run_once()

                except Exception as err:

                    tb_list = async_traceback.extract_tb()