def test_streamer_linecount():
    util.log_to_stderr(util.SUBDEBUG)
    s = Streamer(processes=2)
    stats = s.consume(range(10))
    assert len(stats) == 5
    # 0 is skipped.
    assert stats[PROCESSING_TOTAL] == 10
    assert stats[PROCESSING_SUCCESS] == 9
    assert stats[PROCESSING_SKIPPED] == 1
    assert stats[PROCESSING_ERROR] == 0

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 log_to_stderr(level=None):
    """
    Turn on logging and add a handler which prints to stderr
    """
    from multiprocessing.util import log_to_stderr

    return log_to_stderr(level)

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 setup_logger(loglevel=conf.CELERYD_LOG_LEVEL, logfile=None,
        format=conf.CELERYD_LOG_FORMAT, **kwargs):
    """Setup the ``multiprocessing`` logger. If ``logfile`` is not specified,
    ``stderr`` is used.

    Returns logger object.
    """

    logger = get_default_logger(loglevel=loglevel)
    if logger.handlers:
        # Logger already configured
        return logger
    if logfile:
        handler = logging.FileHandler
        if hasattr(logfile, "write"):
            handler = logging.StreamHandler
        loghandler = handler(logfile)
        formatter = logging.Formatter(format)
        loghandler.setFormatter(formatter)
        logger.addHandler(loghandler)
    else:
        from multiprocessing.util import log_to_stderr
        log_to_stderr()
    return logger

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
        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__':
            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 list(main_module.__dict__.values()):
                try:
                    if obj.__module__ == '__parents_main__':
                        obj.__module__ = '__main__'
                except Exception:
                    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]
        if main_name == '__init__':
            main_name = 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)]

            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

 if os.path.islink(logfile):
     sys.exit('Logfile %s is a symlink. Exiting..' % logfile)
 try:
     logging.basicConfig(filename=logfile, level=level, format='%(asctime)s %(levelname)s %(funcName)s:%(lineno)d %(message)s', datefmt='%Y-%m-%d %H:%M:%S')
     os.chmod(logfile, 0600)
 except IOError as (errno, strerror):
     if errno == int('13'):
         sys.exit('Error while writing to logfile: %s' % strerror)
 try:
     database = Database('yamls/', includes)
     if len(database.issues) == 0:
         sys.exit('Empty database. Exiting..')
     # stderr to /dev/null
     devnull_fd = open(os.devnull, "w")
     sys.stderr = devnull_fd
     log_to_stderr()
     # Starts the asynchronous workers. Amount of workers is the same as cores in server.
     # http://docs.python.org/library/multiprocessing.html#multiprocessing.pool.multiprocessing.Pool
     logging.debug('Starting workers.')
     pool = Pool()
     pool.apply_async(Worker, [arguments['--home'], post_process])
     # Starts the actual populator daemon to get possible locations, which will be verified by workers.
     # http://docs.python.org/library/multiprocessing.html#multiprocessing.Process
     p = PopulateScanQueue()
     if arguments['-r']:
         logging.debug('Scanning recursively from path: %s', arguments['-r'])
         populator = Process(target=p.populate, args=([arguments['-r']],))
     elif arguments['--home']:
         logging.debug('Scanning predefined variables: %s', arguments['--home'])
         populator = Process(target=p.populate_predefined, args=(arguments['--home'], arguments['--check-modes'],))
     else:

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
        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__':
            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
            sys.modules.pop('__mp_main__', None)
            file, path_name, etc = imp.find_module(main_name, dirs)
            try:
                # We should not do 'imp.load_module("__main__", ...)'
                # since that would execute 'if __name__ == "__main__"'
                # clauses, potentially causing a psuedo fork bomb.
                main_module = imp.load_module(
                    '__mp_main__', file, path_name, etc
                    )
            finally:
                if file:
                    file.close()

            sys.modules['__main__'] = sys.modules['__mp_main__'] = main_module

 def parse_l(self, args):
     """ Parse log-level and set it."""
     log_to_stderr(args.loglevel)

if __name__ == "__main__":
    # load node id's that we have already processed
    processed_ids = set()
    try:
        with open('results/processed_similarity_ids.txt') as file_in:
            for line in file_in:
                processed_ids.add(int(line))
    except:
        pass

    # process the rest
    nodes_to_process = [x for x in nodes_for_comparison if x not in processed_ids]

    # parallel generation of results
    util.log_to_stderr(util.SUBDEBUG)
    n_cores = cpu_count()
    chunksize = max(1, len(nodes_to_process) // (2 * (n_cores - 1)))

    
    pool = Pool(processes=(n_cores - 1))
    similarity_results = pool.imap(find_missing_edges,
                                   nodes_to_process,
                                   chunksize=chunksize)

    # write results to file
    with open('results/similarity_results.csv', 'a') as file_out, \
         open('results/processed_similarity_ids.txt', 'a') as processed:
        for result in similarity_results:
            if not result:
                continue

    def __init__(self, processes=None, initializer=None, initargs=(),
                 maxtasksperchild=None, timeout_seconds=30):
        log_to_stderr(level=DEBUG)

        self._setup_queues()
        self._taskqueue = Queue.Queue()
        self._cache = {}
        self._state = RUN
        self._maxtasksperchild = maxtasksperchild
        self._initializer = initializer
        self._initargs = initargs
        self._proc_num = 0

        if processes is None:
            try:
                processes = cpu_count()
            except NotImplementedError:
                processes = 1

        if initializer is not None and not hasattr(initializer, '__call__'):
            raise TypeError('initializer must be a callable')

        self._processes = processes
        self._pool = []
        self._repopulate_pool()

        self._worker_handler = threading.Thread(
            target=Pool._handle_workers,
            args=(self, )
        )

        self._worker_handler.daemon = True
        self._worker_handler._state = RUN
        self._worker_handler.start()

        self._task_handler = threading.Thread(
            target=Pool._handle_tasks,
            args=(self._taskqueue, self._quick_put, self._outqueue, self._pool)
        )

        self._task_handler.daemon = True
        self._task_handler._state = RUN
        self._task_handler.start()

        self.timeout_seconds = timeout_seconds

        self._result_handler = threading.Thread(
            target=TimeoutPool._handle_results,
            args=(self._outqueue, self._quick_get, self._cache, self._pool, self.timeout_seconds)
        )

        self._result_handler.daemon = True
        self._result_handler._state = RUN
        self._result_handler.start()

        self._terminate = Finalize(
            self, self._terminate_pool,
            args=(self._taskqueue, self._inqueue, self._outqueue, self._pool,
                  self._worker_handler, self._task_handler,
                  self._result_handler, self._cache),
            exitpriority=15
        )

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