def __init__(self, socket_filename, max_connections=64,
                 connection_timeout=10, poll_timeout=1, num_workers=1,
                 num_worker_processes=2, max_processed_jobs=1,
                 max_queued_jobs=8, logger=None):
        self.socket_address = socket_filename
        self.socket_family = 'AF_UNIX'
        self.socket_kwargs = {}
        self.socket_listener = None
        self.connection_semaphore = Semaphore(max_connections)
        self.connection_handlers = ThreadSet()
        self.connection_timeout = connection_timeout
        self.connection_poll_timeout = poll_timeout
        self.job_queue = Queue(max_queued_jobs)

        self.logger = logger or getLogger(LOGGER_NAME)

        # set verbose multi-processes debugging
        set_stream_handler(mp_util.get_logger())
        mp_util.get_logger().setLevel(mp_util.SUBWARNING)
        worker_semaphore = ProcessSemaphore(num_workers)

        self.worker_pool = ProcessPool(
            max(num_workers, num_worker_processes), init_worker,
            maxtasksperchild=max_processed_jobs,
        )
        self.worker_pool_manager = WorkerPoolManager(
            self.worker_pool, worker_semaphore, self.job_queue
        )

def teardown():
    # Don't want SUBDEBUG log messages at finalization.
    try:
        from multiprocessing.util import get_logger
    except ImportError:
        pass
    else:
        get_logger().setLevel(logging.WARNING)

    # Make sure test database is removed.
    import os
    if os.path.exists('test.db'):
        try:
            os.remove('test.db')
        except WindowsError:
            pass

    # Make sure there are no remaining threads at shutdown.
    import threading
    remaining_threads = [thread for thread in threading.enumerate()
                         if thread.getName() != 'MainThread']
    if remaining_threads:
        sys.stderr.write(
            '\n\n**WARNING**: Remaining threads at teardown: %r...\n' % (
                remaining_threads))

def start_command(config, args):
    initialize_logger(config)
    print('starting command: %s' % ' '.join(args))
    get_logger().info("Starting command: %s" % " ".join(args))
    try:
        subprocess.call(args, shell=True)
    except:
        pass

def prepare(data):
    """
    Try to get current process ready to unpickle process object
    """
    old_main_modules.append(sys.modules['__main__'])
    if 'name' in data:
        process.current_process().name = data['name']
    if 'authkey' in data:
        process.current_process()._authkey = data['authkey']
    if 'log_to_stderr' in data and data['log_to_stderr']:
        util.log_to_stderr()
    if 'log_level' in data:
        util.get_logger().setLevel(data['log_level'])
    if 'sys_path' in data:
        sys.path = data['sys_path']
    if 'sys_argv' in data:
        sys.argv = data['sys_argv']
    if 'dir' in data:
        os.chdir(data['dir'])
    if 'orig_dir' in data:
        process.ORIGINAL_DIR = data['orig_dir']
    if 'main_path' in data:
        main_path = data['main_path']
        main_name = os.path.splitext(os.path.basename(main_path))[0]
        main_name = main_name == '__init__' and os.path.basename(os.path.dirname(main_path))
    if main_name != 'ipython':
        import imp
        if main_path is None:
            dirs = None
        elif os.path.basename(main_path).startswith('__init__.py'):
            dirs = [os.path.dirname(os.path.dirname(main_path))]
        else:
            dirs = [os.path.dirname(main_path)]
        if not main_name not in sys.modules:
            raise AssertionError(main_name)
            file, path_name, etc = imp.find_module(main_name, dirs)
            try:
                main_module = imp.load_module('__parents_main__', file, path_name, etc)
            finally:
                if file:
                    file.close()

            sys.modules['__main__'] = main_module
            main_module.__name__ = '__main__'
            for obj in main_module.__dict__.values():
                try:
                    if obj.__module__ == '__parents_main__':
                        obj.__module__ = '__main__'
                except Exception:
                    pass

    return

def teardown():
    # Don't want SUBDEBUG log messages at finalization.
    from multiprocessing.util import get_logger
    get_logger().setLevel(logging.WARNING)
    import threading
    import os
    if os.path.exists("test.db"):
        os.remove("test.db")
    remaining_threads = [thread for thread in threading.enumerate()
                            if thread.name != "MainThread"]
    if remaining_threads:
        sys.stderr.write(
            "\n\n**WARNING**: Remaning threads at teardown: %r...\n" % (
                remaining_threads))

    def setup_logging_subsystem(self, loglevel=None, logfile=None,
            format=None, colorize=None, **kwargs):
        loglevel = loglevel or self.loglevel
        format = format or self.format
        colorize = self.app.either("CELERYD_LOG_COLOR", colorize)

        if self.__class__._setup:
            return

        try:
            mputil._logger = None
        except AttributeError:
            pass
        ensure_process_aware_logger()
        logging.Logger.manager.loggerDict.clear()
        receivers = signals.setup_logging.send(sender=None,
                                               loglevel=loglevel,
                                               logfile=logfile,
                                               format=format,
                                               colorize=colorize)
        if not receivers:
            root = logging.getLogger()
            mp = mputil.get_logger()
            for logger in (root, mp):
                self._setup_logger(logger, logfile,
                                   format, colorize, **kwargs)
                logger.setLevel(loglevel)
        self.__class__._setup = True
        return receivers

    def setup_logging_subsystem(self, loglevel=None, logfile=None, format=None, colorize=None, **kwargs):
        if Logging._setup:
            return
        loglevel = mlevel(loglevel or self.loglevel)
        format = format or self.format
        if colorize is None:
            colorize = self.supports_color(logfile)

        if mputil and hasattr(mputil, "_logger"):
            mputil._logger = None
        if not is_py3k:
            ensure_process_aware_logger()
        receivers = signals.setup_logging.send(
            sender=None, loglevel=loglevel, logfile=logfile, format=format, colorize=colorize
        )
        if not receivers:
            root = logging.getLogger()

            if self.app.conf.CELERYD_HIJACK_ROOT_LOGGER:
                root.handlers = []

            mp = mputil.get_logger() if mputil else None
            for logger in filter(None, (root, mp)):
                self._setup_logger(logger, logfile, format, colorize, **kwargs)
                logger.setLevel(mlevel(loglevel))
                signals.after_setup_logger.send(
                    sender=None, logger=logger, loglevel=loglevel, logfile=logfile, format=format, colorize=colorize
                )

        # This is a hack for multiprocessing's fork+exec, so that
        # logging before Process.run works.
        os.environ.update(_MP_FORK_LOGLEVEL_=str(loglevel), _MP_FORK_LOGFILE_=logfile or "", _MP_FORK_LOGFORMAT_=format)
        Logging._setup = True

        return receivers

def get_logger():
    """
    Return package logger -- if it does not already exist then it is created
    """
    from multiprocessing.util import get_logger

    return get_logger()

    def setup_logging_subsystem(self, loglevel=None, logfile=None,
            format=None, colorize=None, **kwargs):
        if Logging._setup:
            return
        loglevel = loglevel or self.loglevel
        format = format or self.format
        if colorize is None:
            colorize = self.supports_color(logfile)

        if mputil and hasattr(mputil, "_logger"):
            mputil._logger = None
        ensure_process_aware_logger()
        receivers = signals.setup_logging.send(sender=None,
                        loglevel=loglevel, logfile=logfile,
                        format=format, colorize=colorize)
        if not receivers:
            root = logging.getLogger()

            if self.app.conf.CELERYD_HIJACK_ROOT_LOGGER:
                root.handlers = []

            mp = mputil.get_logger() if mputil else None
            for logger in filter(None, (root, mp)):
                self._setup_logger(logger, logfile, format, colorize, **kwargs)
                logger.setLevel(loglevel)
                signals.after_setup_logger.send(sender=None, logger=logger,
                                        loglevel=loglevel, logfile=logfile,
                                        format=format, colorize=colorize)
        Logging._setup = True

        return receivers

    def main():
        assert is_forking(sys.argv)
        fd = int(sys.argv[-1])
        from_parent = os.fdopen(fd, 'rb')
        current_process()._inheriting = True
        preparation_data = load(from_parent)
        _forking.prepare(preparation_data)

        # Huge hack to make logging before Process.run work.
        loglevel = os.environ.get("_MP_FORK_LOGLEVEL_")
        logfile = os.environ.get("_MP_FORK_LOGFILE_") or None
        format = os.environ.get("_MP_FORK_LOGFORMAT_")
        if loglevel:
            from multiprocessing import util
            import logging
            logger = util.get_logger()
            logger.setLevel(int(loglevel))
            if not logger.handlers:
                logger._rudimentary_setup = True
                logfile = logfile or sys.__stderr__
                if hasattr(logfile, "write"):
                    handler = logging.StreamHandler(logfile)
                else:
                    handler = logging.FileHandler(logfile)
                formatter = logging.Formatter(
                        format or util.DEFAULT_LOGGING_FORMAT)
                handler.setFormatter(formatter)
                logger.addHandler(handler)

        self = load(from_parent)
        current_process()._inheriting = False

        exitcode = self._bootstrap()
        exit(exitcode)

def setup_logging_subsystem(loglevel=conf.CELERYD_LOG_LEVEL, logfile=None,
        format=conf.CELERYD_LOG_FORMAT, colorize=conf.CELERYD_LOG_COLOR,
        **kwargs):
    global _setup
    if not _setup:
        try:
            mputil._logger = None
        except AttributeError:
            pass
        ensure_process_aware_logger()
        logging.Logger.manager.loggerDict.clear()
        receivers = signals.setup_logging.send(sender=None,
                                               loglevel=loglevel,
                                               logfile=logfile,
                                               format=format,
                                               colorize=colorize)
        if not receivers:
            root = logging.getLogger()

            if conf.CELERYD_HIJACK_ROOT_LOGGER:
                root.handlers = []

            mp = mputil.get_logger()
            for logger in (root, mp):
                _setup_logger(logger, logfile, format, colorize, **kwargs)
                logger.setLevel(loglevel)
        _setup = True
        return receivers

    def setup_logging_subsystem(self, loglevel=None, logfile=None,
            format=None, colorize=None, **kwargs):
        if Logging._setup:
            return
        loglevel = loglevel or self.loglevel
        format = format or self.format
        if colorize is None:
            colorize = self.supports_color(logfile)

        if mputil:
            try:
                mputil._logger = None
            except AttributeError:
                pass
        ensure_process_aware_logger()
        receivers = signals.setup_logging.send(sender=None,
                                               loglevel=loglevel,
                                               logfile=logfile,
                                               format=format,
                                               colorize=colorize)
        if not receivers:
            root = logging.getLogger()

            if self.app.conf.CELERYD_HIJACK_ROOT_LOGGER:
                root.handlers = []

            mp = mputil and mputil.get_logger() or None
            for logger in (root, mp):
                if logger:
                    self._setup_logger(logger, logfile, format,
                                       colorize, **kwargs)
                    logger.setLevel(loglevel)
        Logging._setup = True
        return receivers

def initialize_logger(config):
    class StdErrWrapper:
        """
            Call wrapper for stderr
        """
        def write(self, s):
            get_logger().info(s)
    import logging

    logger = get_logger()
    values = dict(
        format='[%(levelname)s/%(processName)s] %(message)s',
        filename=None,
        level='INFO',
    )
    if config and config.has_section('log'):
        for (name, value) in config.items('log'):
            values[name] = value

    if values['filename']:
        formatter = logging.Formatter(values['format'])
        handler = logging.FileHandler(values['filename'])
        handler.setFormatter(formatter)
        logger.addHandler(handler)
        logger.setLevel(
            getattr(logging, values['level'].upper(), logging.INFO)
        )
        sys.stderr = StdErrWrapper()

def start_django_command(config, args):
    '''
    Start a Django management command.
    
    This commands is supposed to start in a spawned (child process).
    It tries to import the settings of the project before handling the command.
    '''    
    initialize_logger(config)

    log('Starting command : %s' % ' '.join(args))
    get_logger().info('Starting command : %s' % ' '.join(args))
    from django.core.management import execute_from_command_line
    
    try:
        execute_from_command_line(args)
    except:
        error('Exception occured : %s' % traceback.format_exc())

def prepare(data):
    '''
    Try to get current process ready to unpickle process object
    '''
    if 'name' in data:
        process.current_process().name = data['name']

    if 'authkey' in data:
        process.current_process().authkey = data['authkey']

    if 'log_to_stderr' in data and data['log_to_stderr']:
        util.log_to_stderr()

    if 'log_level' in data:
        util.get_logger().setLevel(data['log_level'])

    if 'log_fmt' in data:
        import logging
        util.get_logger().handlers[0].setFormatter(
            logging.Formatter(data['log_fmt'])
        )

    if 'sys_path' in data:
        sys.path = data['sys_path']

    if 'sys_argv' in data:
        sys.argv = data['sys_argv']

    if 'dir' in data:
        os.chdir(data['dir'])

    if 'orig_dir' in data:
        process.ORIGINAL_DIR = data['orig_dir']

    if hasattr(mp, 'set_start_method'):
        mp.set_start_method('loky', force=True)

    if 'tacker_pid' in data:
        from . import semaphore_tracker
        semaphore_tracker._semaphore_tracker._pid = data["tracker_pid"]

    if 'init_main_from_name' in data:
        _fixup_main_from_name(data['init_main_from_name'])
    elif 'init_main_from_path' in data:
        _fixup_main_from_path(data['init_main_from_path'])

def _hijack_multiprocessing_logger():
    from multiprocessing import util as mputil
    global _hijacked

    if _hijacked:
        return mputil.get_logger()

    ensure_process_aware_logger()

    logging.Logger.manager.loggerDict.clear()

    try:
        if mputil._logger is not None:
            mputil.logger = None
    except AttributeError:
        pass

    _hijacked = True
    return mputil.get_logger()

def get_default_logger(loglevel=None):
    """Get default logger instance.

    :keyword loglevel: Initial log level.

    """
    from multiprocessing.util import get_logger
    logger = get_logger()
    loglevel is not None and logger.setLevel(loglevel)
    return logger

def teardown():
    # Don't want SUBDEBUG log messages at finalization.
    from multiprocessing.util import get_logger
    get_logger().setLevel(logging.WARNING)
    import threading
    import os
    if os.path.exists("test.db"):
        os.remove("test.db")
    remaining_threads = [thread for thread in threading.enumerate()
                            if thread.name != "MainThread"]
    if remaining_threads:
        sys.stderr.write(
            "\n\n**WARNING**: Remaning threads at teardown: %r...\n" % (
                remaining_threads))
        for thread in remaining_threads:
            try:
                started_by = thread._started_by[thread.ident]
            except (AttributeError, KeyError):
                pass
            else:
                sys.stderr.write("THREAD %r STARTED BY:\n%r\n" % (
                    thread, started_by))

def prepare(data):
    """
    Try to get current process ready to unpickle process object
    """
    old_main_modules.append(sys.modules["__main__"])

    if "name" in data:
        process.current_process().name = data["name"]

    if "authkey" in data:
        process.current_process()._authkey = data["authkey"]

    if "log_to_stderr" in data and data["log_to_stderr"]:
        util.log_to_stderr()

    if "log_level" in data:
        util.get_logger().setLevel(data["log_level"])

    if "sys_path" in data:
        sys.path = data["sys_path"]

    if "sys_argv" in data:
        sys.argv = data["sys_argv"]

    if "dir" in data:
        os.chdir(data["dir"])

    if "orig_dir" in data:
        process.ORIGINAL_DIR = data["orig_dir"]

    if "main_path" in data:
        # XXX (ncoghlan): The following code makes several bogus
        # assumptions regarding the relationship between __file__
        # and a module's real name. See PEP 302 and issue #10845
        # The problem is resolved properly in Python 3.4+, as
        # described in issue #19946

        main_path = data["main_path"]
        main_name = os.path.splitext(os.path.basename(main_path))[0]
        if main_name == "__init__":
            main_name = os.path.basename(os.path.dirname(main_path))

        if main_name == "__main__":
            # For directory and zipfile execution, we assume an implicit
            # "if __name__ == '__main__':" around the module, and don't
            # rerun the main module code in spawned processes
            main_module = sys.modules["__main__"]
            main_module.__file__ = main_path
        elif main_name != "ipython":
            # Main modules not actually called __main__.py may
            # contain additional code that should still be executed
            import imp

            if main_path is None:
                dirs = None
            elif os.path.basename(main_path).startswith("__init__.py"):
                dirs = [os.path.dirname(os.path.dirname(main_path))]
            else:
                dirs = [os.path.dirname(main_path)]

            assert main_name not in sys.modules, main_name
            file, path_name, etc = imp.find_module(main_name, dirs)
            try:
                # We would like to do "imp.load_module('__main__', ...)"
                # here.  However, that would cause 'if __name__ ==
                # "__main__"' clauses to be executed.
                main_module = imp.load_module("__parents_main__", file, path_name, etc)
            finally:
                if file:
                    file.close()

            sys.modules["__main__"] = main_module
            main_module.__name__ = "__main__"

            # Try to make the potentially picklable objects in
            # sys.modules['__main__'] realize they are in the main
            # module -- somewhat ugly.
            for obj in main_module.__dict__.values():
                try:
                    if obj.__module__ == "__parents_main__":
                        obj.__module__ = "__main__"
                except Exception:
                    pass

 def __init__(self, limit, logger=None):
     self.limit = limit
     self.logger = logger or get_logger()
     self._pool = None