def _actor_loop(self):
        """
        The actor's event loop.

        This is the method that will be executed by the thread or greenlet.
        """
        self.on_start()
        while self._actor_runnable:
            message = self.actor_inbox.get()
            try:
                response = self._handle_receive(message)
                if 'reply_to' in message:
                    message['reply_to'].set(response)
            except Exception:
                if 'reply_to' in message:
                    _logger.debug(
                        'Exception returned from %s to caller:' % self,
                        exc_info=_sys.exc_info())
                    message['reply_to'].set_exception()
                else:
                    self._handle_failure(*_sys.exc_info())
            except BaseException:
                exception_value = _sys.exc_info()[1]
                _logger.debug(
                    '%s in %s. Stopping all actors.' %
                    (repr(exception_value), self))
                self._stop()
                _ActorRegistry.stop_all()

 def _handle_failure(self, exception_type, exception_value, traceback):
     """Logs unexpected failures, unregisters and stops the actor."""
     logger.error(
         'Unhandled exception in %s:' % self,
         exc_info=(exception_type, exception_value, traceback))
     ActorRegistry.unregister(self.actor_ref)
     self.actor_stopped.set()

def scan_command(directory, tagfile='STAG.sqlite', verbose=False):
    """Scan a directory tree for definitions and references."""

    init_logging(verbose)

    with StorageManager(Storage(tagfile)) as s:
        s.clear_defs()

        storage = StorageActor.start(s)
        parser_map = [
            (list(p.patterns()),
             ParserActor.start(p.create_parser(), storage))
            for p in parser_plugins()]

        dispatcher = DispatcherActor.start(parser_map)

        def dispatch_file(args):
            "Send results of os.walk to the dispatcher."
            dirpath, _, filenames = args
            for fname in filenames:
                dispatcher.tell({
                    'command': 'dispatch',
                    'filename': os.path.join(dirpath, fname)
        })

        emap(dispatch_file, os.walk(directory))

        # Shut everything down.
        ActorRegistry.stop_all()

 def _handle_failure(self, exception_type, exception_value, traceback):
     """Logs unexpected failures, unregisters and stops the actor."""
     _logger.error('Unhandled exception in %s:' % self,
         exc_info=(exception_type, exception_value, traceback))
     _ActorRegistry.unregister(self.actor_ref)
     self._actor_runnable = False
     self.on_failure(exception_type, exception_value, traceback)

    def _run(self):
        """
        The actor's main method.

        :class:`pykka.gevent.GeventActor` expects this method to be named
        :meth:`_run`.

        :class:`ThreadingActor` expects this method to be named :meth:`run`.
        """
        self.on_start()
        while self._actor_runnable:
            try:
                message = self.actor_inbox.get(False)
                _logger.debug('Actor {} got message {}'.format(self, message))
                try:
                    response = self._handle_receive(message)
                    if 'reply_to' in message:
                        message['reply_to'].set(response)
                except Exception as exception:
                    if 'reply_to' in message:
                        _logger.debug('Exception returned from %s to caller:' %
                            self, exc_info=_sys.exc_info())
                        message['reply_to'].set_exception(exception)
                    else:
                        self._handle_failure(*_sys.exc_info())
                except BaseException as exception:
                    exception_value = _sys.exc_info()[1]
                    _logger.debug('%s in %s. Stopping all actors.' %
                        (repr(exception_value), self))
                    self.stop()
                    _ActorRegistry.stop_all()
            except _queue.Empty:
                pass

    def start(cls, *args, **kwargs):
        """
        Start an actor and register it in the
        :class:`pykka.registry.ActorRegistry`.

        Any arguments passed to :meth:`start` will be passed on to the class
        constructor.

        Returns a :class:`ActorRef` which can be used to access the actor in a
        safe manner.

        Behind the scenes, the following is happening when you call
        :meth:`start`::

            Actor.start()
                Actor.__new__()
                    superclass.__new__()
                    superclass.__init__()
                    URN assignment
                    Inbox creation
                    ActorRef creation
                Actor.__init__()        # Your code can run here
                ActorRegistry.register()
                superclass.start()
        """
        obj = cls(*args, **kwargs)
        _ActorRegistry.register(obj.actor_ref)
        cls._superclass.start(obj)
        _logger.debug('Started %s', obj)
        return obj.actor_ref

 def _stop(self):
     """
     Stops the actor immediately without processing the rest of the inbox.
     """
     _ActorRegistry.unregister(self.actor_ref)
     self._actor_runnable = False
     _logger.debug('Stopped %s', self)
     self.on_stop()

 def test_broadcast_sends_message_to_all_actors_of_given_class_name(self):
     ActorRegistry.broadcast({'command': 'foo'}, target_class='AnActor')
     for actor_ref in ActorRegistry.get_by_class(self.AnActor):
         received_messages = actor_ref.proxy().received_messages.get()
         self.assert_({'command': 'foo'} in received_messages)
     for actor_ref in ActorRegistry.get_by_class(self.BeeActor):
         received_messages = actor_ref.proxy().received_messages.get()
         self.assert_({'command': 'foo'} not in received_messages)

    def setup(self):
        gstreamer_refs = ActorRegistry.get_by_class(GStreamer)
        assert len(gstreamer_refs) == 1, \
            'Expected exactly one running gstreamer.'
        self.gstreamer = gstreamer_refs[0].proxy()

        backend_refs = ActorRegistry.get_by_class(Backend)
        assert len(backend_refs) == 1, 'Expected exactly one running backend.'
        self.backend = backend_refs[0].proxy()

    def setup(self):
        audio_refs = ActorRegistry.get_by_class(audio.Audio)
        assert len(audio_refs) == 1, \
            'Expected exactly one running Audio instance.'
        self.audio = audio_refs[0].proxy()

        backend_refs = ActorRegistry.get_by_class(Backend)
        assert len(backend_refs) == 1, 'Expected exactly one running backend.'
        self.backend = backend_refs[0].proxy()

def stop_all_actors():
    num_actors = len(ActorRegistry.get_all())
    while num_actors:
        logger.debug(u'Seeing %d actor and %d non-actor thread(s): %s',
            num_actors, threading.active_count() - num_actors,
            ', '.join([t.name for t in threading.enumerate()]))
        logger.debug(u'Stopping %d actor(s)...', num_actors)
        ActorRegistry.stop_all()
        num_actors = len(ActorRegistry.get_all())
    logger.debug(u'All actors stopped.')

 def stop(self):
     logger.info("exiting ...")
     logger.info("waiting for actors to stop ...")
     try:
         ActorRegistry.stop_all(timeout=10)
     except Exception as ex:
         logger.info("warning - actors failed to stop cleanly")
     logger.info("stopping web server ...")
     stop_web()
     logger.info("finished")

    def stop(self):
        """
        Stop the actor.

        The actor will finish processing any messages already in its queue
        before stopping. It may not be restarted.
        """
        self.actor_runnable = False
        ActorRegistry.unregister(self.actor_ref)
        logger.debug(u'Stopped %s', self)

 def test_actor_may_be_unregistered_multiple_times_without_error(self):
     ActorRegistry.unregister(self.ref)
     self.assert_(self.ref not in ActorRegistry.get_all())
     ActorRegistry.unregister(self.ref)
     self.assert_(self.ref not in ActorRegistry.get_all())
     ActorRegistry.register(self.ref)
     self.assert_(self.ref in ActorRegistry.get_all())

 def _stop(self):
     """
     Stops the actor immediately without processing the rest of the inbox.
     """
     _ActorRegistry.unregister(self.actor_ref)
     self._actor_runnable = False
     _logger.debug('Stopped %s', self)
     try:
         self.on_stop()
     except Exception:
         self._handle_failure(*_sys.exc_info())

def stop_remaining_actors():
    num_actors = len(ActorRegistry.get_all())
    while num_actors:
        logger.error(
            u'There are actor threads still running, this is probably a bug')
        logger.debug(u'Seeing %d actor and %d non-actor thread(s): %s',
            num_actors, threading.active_count() - num_actors,
            ', '.join([t.name for t in threading.enumerate()]))
        logger.debug(u'Stopping %d actor(s)...', num_actors)
        ActorRegistry.stop_all()
        num_actors = len(ActorRegistry.get_all())
    logger.debug(u'All actors stopped.')

 def test_runs_parallel(self):
     self.actor = CoroutineActor.start().proxy()
     first = self.actor.infinite_loop_count.get()
     self.actor.loop_infinitely(0).get()
     second = self.actor.infinite_loop_count.get()
     self.actor.other_method().get()
     called = self.actor.other_called.get()
     third = self.actor.infinite_loop_count.get()
     self.assertTrue(called)
     self.assertLess(first, second)
     self.assertLess(second, third)
     ActorRegistry.stop_all()

    def _actor_loop(self):
        """
        The actor's event loop.

        This is the method that will be executed by the thread or greenlet.
        """
        try:
            self.on_start()
        except Exception:
            self._handle_failure(*sys.exc_info())

        while not self.actor_stopped.is_set():
            message = self.actor_inbox.get()
            reply_to = None
            try:
                reply_to = message.pop('pykka_reply_to', None)
                response = self._handle_receive(message)
                if reply_to:
                    reply_to.set(response)
            except Exception:
                if reply_to:
                    logger.debug(
                        'Exception returned from %s to caller:' % self,
                        exc_info=sys.exc_info())
                    reply_to.set_exception()
                else:
                    self._handle_failure(*sys.exc_info())
                    try:
                        self.on_failure(*sys.exc_info())
                    except Exception:
                        self._handle_failure(*sys.exc_info())
            except BaseException:
                exception_value = sys.exc_info()[1]
                logger.debug(
                    '%s in %s. Stopping all actors.' %
                    (repr(exception_value), self))
                self._stop()
                ActorRegistry.stop_all()

        while not self.actor_inbox.empty():
            msg = self.actor_inbox.get()
            reply_to = msg.pop('pykka_reply_to', None)
            if reply_to:
                if msg.get('command') == 'pykka_stop':
                    reply_to.set(None)
                else:
                    reply_to.set_exception(ActorDeadError(
                        '%s stopped before handling the message' %
                        self.actor_ref))

def run(pool_size, *ips):
    # Start resolvers
    resolvers = [Resolver.start_proxy() for _ in range(pool_size)]

    # Distribute work by mapping IPs to resolvers (not blocking)
    hosts = []
    for i, ip in enumerate(ips):
        hosts.append(resolvers[i % len(resolvers)].resolve(ip))

    # Gather results (blocking)
    ip_to_host = zip(ips, get_all(hosts))
    pprint(ip_to_host)

    # Clean up
    ActorRegistry.stop_all()

    def test_all_actors_are_stopped_on_base_exception(self):
        start_event = self.event_class()
        stop_event = self.event_class()
        fail_event = self.event_class()
        registered_event = self.event_class()
        another_actor = self.AnActor.start(start_event, stop_event, fail_event,
            registered_event)

        self.assertEqual(2, len(ActorRegistry.get_all()))
        self.assertFalse(self.on_stop_was_called.is_set())
        self.actor_ref.send_one_way({'command': 'raise base exception'})
        self.on_stop_was_called.wait()
        self.assert_(1 >= len(ActorRegistry.get_all()))
        stop_event.wait()
        self.assertEqual(0, len(ActorRegistry.get_all()))