def test_double_reader_writer(self):
        lock = lock_utils.ReaderWriterLock()
        activated = collections.deque()
        active = threading_utils.Event()

        def double_reader():
            with lock.read_lock():
                active.set()
                while not lock.has_pending_writers:
                    time.sleep(0.001)
                with lock.read_lock():
                    activated.append(lock.owner)

        def happy_writer():
            with lock.write_lock():
                activated.append(lock.owner)

        reader = threading_utils.daemon_thread(double_reader)
        reader.start()
        self.assertTrue(active.wait(test_utils.WAIT_TIMEOUT))

        writer = threading_utils.daemon_thread(happy_writer)
        writer.start()

        reader.join()
        writer.join()
        self.assertEqual(2, len(activated))
        self.assertEqual(['r', 'w'], list(activated))

    def test_wait_arrival(self):
        ev = threading.Event()
        jobs = []

        def poster(wait_post=0.2):
            if not ev.wait(test_utils.WAIT_TIMEOUT):
                raise RuntimeError("Waiter did not appear ready"
                                   " in %s seconds" % test_utils.WAIT_TIMEOUT)
            time.sleep(wait_post)
            self.board.post('test', p_utils.temporary_log_book())

        def waiter():
            ev.set()
            it = self.board.wait()
            jobs.extend(it)

        with connect_close(self.board):
            t1 = threading_utils.daemon_thread(poster)
            t1.start()
            t2 = threading_utils.daemon_thread(waiter)
            t2.start()
            for t in (t1, t2):
                t.join()

        self.assertEqual(1, len(jobs))

def main():
    if six.PY3:
        # TODO(harlowja): Hack to make eventlet work right, remove when the
        # following is fixed: https://github.com/eventlet/eventlet/issues/230
        from taskflow.utils import eventlet_utils as _eu  # noqa
        try:
            import eventlet as _eventlet  # noqa
        except ImportError:
            pass
    with contextlib.closing(fake_client.FakeClient()) as c:
        created = []
        for i in compat_range(0, PRODUCERS):
            p = threading_utils.daemon_thread(producer, i + 1, c)
            created.append(p)
            p.start()
        consumed = collections.deque()
        for i in compat_range(0, WORKERS):
            w = threading_utils.daemon_thread(worker, i + 1, c, consumed)
            created.append(w)
            w.start()
        while created:
            t = created.pop()
            t.join()
        # At the end there should be nothing leftover, let's verify that.
        board = backends.fetch('verifier', SHARED_CONF.copy(), client=c)
        board.connect()
        with contextlib.closing(board):
            if board.job_count != 0 or len(consumed) != EXPECTED_UNITS:
                return 1
            return 0

 def start(self):
     """Starts proxy thread and associated topic notification thread."""
     if not _is_alive(self._proxy_thread):
         self._proxy_thread = tu.daemon_thread(self._proxy.start)
         self._proxy_thread.start()
         self._proxy.wait()
     if not _is_alive(self._periodic_thread):
         self._periodic.reset()
         self._periodic_thread = tu.daemon_thread(self._periodic.start)
         self._periodic_thread.start()

    def test_double_acquire_many(self):
        activated = collections.deque()
        n_lock = lock_utils.MultiLock((threading.RLock(), threading.RLock()))

        def critical_section():
            start = time.time()
            time.sleep(0.05)
            end = time.time()
            activated.append((start, end))

        def run():
            with n_lock:
                critical_section()
                with n_lock:
                    critical_section()
                critical_section()

        threads = []
        for i in range(0, 20):
            t = threading_utils.daemon_thread(run)
            threads.append(t)
            t.start()
        while threads:
            t = threads.pop()
            t.join()

        for (start, end) in activated:
            self.assertEqual(1, _find_overlaps(activated, start, end))

    def test_acquired_pass(self):
        activated = collections.deque()
        lock1 = threading.Lock()
        lock2 = threading.Lock()
        n_lock = lock_utils.MultiLock((lock1, lock2))

        def critical_section():
            start = time.time()
            time.sleep(0.05)
            end = time.time()
            activated.append((start, end))

        def run():
            with n_lock:
                critical_section()

        threads = []
        for _i in range(0, 20):
            t = threading_utils.daemon_thread(run)
            threads.append(t)
            t.start()
        while threads:
            t = threads.pop()
            t.join()
        for (start, end) in activated:
            self.assertEqual(1, _find_overlaps(activated, start, end))

        self.assertFalse(lock1.locked())
        self.assertFalse(lock2.locked())

    def test_combined_store(self):
        components = self.make_components()
        components.conductor.connect()
        consumed_event = threading.Event()

        def on_consume(state, details):
            consumed_event.set()

        flow_store = {'x': True, 'y': False}
        job_store = {'z': None}

        components.board.notifier.register(base.REMOVAL, on_consume)
        with close_many(components.conductor, components.client):
            t = threading_utils.daemon_thread(components.conductor.run)
            t.start()
            lb, fd = pu.temporary_flow_detail(components.persistence,
                                              meta={'store': flow_store})
            engines.save_factory_details(fd, test_store_factory,
                                         [], {},
                                         backend=components.persistence)
            components.board.post('poke', lb,
                                  details={'flow_uuid': fd.uuid,
                                           'store': job_store})
            self.assertTrue(consumed_event.wait(test_utils.WAIT_TIMEOUT))
            components.conductor.stop()
            self.assertTrue(components.conductor.wait(test_utils.WAIT_TIMEOUT))
            self.assertFalse(components.conductor.dispatching)

        persistence = components.persistence
        with contextlib.closing(persistence.get_connection()) as conn:
            lb = conn.get_logbook(lb.uuid)
            fd = lb.find(fd.uuid)
        self.assertIsNotNone(fd)
        self.assertEqual(st.SUCCESS, fd.state)

    def test_response(self):
        barrier = threading_utils.Event()

        on_response = mock.MagicMock()
        on_response.side_effect = lambda *args, **kwargs: barrier.set()

        handlers = {pr.RESPONSE: dispatcher.Handler(on_response)}
        p = proxy.Proxy(TEST_TOPIC, TEST_EXCHANGE, handlers,
                        transport='memory',
                        transport_options={
                            'polling_interval': POLLING_INTERVAL,
                        })

        t = threading_utils.daemon_thread(p.start)
        t.start()
        p.wait()
        resp = pr.Response(pr.RUNNING)
        p.publish(resp, TEST_TOPIC)

        self.assertTrue(barrier.wait(test_utils.WAIT_TIMEOUT))
        self.assertTrue(barrier.is_set())
        p.stop()
        t.join()

        self.assertTrue(on_response.called)
        on_response.assert_called_with(resp.to_dict(), mock.ANY)

    def test_fail_run(self):
        components = self.make_components()
        components.conductor.connect()
        consumed_event = threading_utils.Event()

        def on_consume(state, details):
            consumed_event.set()

        components.board.notifier.register(jobboard.REMOVAL, on_consume)
        with close_many(components.conductor, components.client):
            t = threading_utils.daemon_thread(components.conductor.run)
            t.start()
            lb, fd = pu.temporary_flow_detail(components.persistence)
            engines.save_factory_details(fd, test_factory,
                                         [True], {},
                                         backend=components.persistence)
            components.board.post('poke', lb,
                                  details={'flow_uuid': fd.uuid})
            self.assertTrue(consumed_event.wait(test_utils.WAIT_TIMEOUT))
            self.assertTrue(components.conductor.stop(test_utils.WAIT_TIMEOUT))
            self.assertFalse(components.conductor.dispatching)

        persistence = components.persistence
        with contextlib.closing(persistence.get_connection()) as conn:
            lb = conn.get_logbook(lb.uuid)
            fd = lb.find(fd.uuid)
        self.assertIsNotNone(fd)
        self.assertEqual(st.REVERTED, fd.state)

    def test_no_double_writers(self):
        lock = lock_utils.ReaderWriterLock()
        watch = timing.StopWatch(duration=5)
        watch.start()
        dups = collections.deque()
        active = collections.deque()

        def acquire_check(me):
            with lock.write_lock():
                if len(active) >= 1:
                    dups.append(me)
                    dups.extend(active)
                active.append(me)
                try:
                    time.sleep(random.random() / 100)
                finally:
                    active.remove(me)

        def run():
            me = threading.current_thread()
            while not watch.expired():
                acquire_check(me)

        threads = []
        for i in range(0, self.THREAD_COUNT):
            t = threading_utils.daemon_thread(run)
            threads.append(t)
            t.start()
        while threads:
            t = threads.pop()
            t.join()

        self.assertEqual([], list(dups))
        self.assertEqual([], list(active))

    def test_run_max_dispatches(self):
        components = self.make_components()
        components.conductor.connect()
        consumed_event = threading.Event()

        def on_consume(state, details):
            consumed_event.set()

        components.board.notifier.register(base.REMOVAL, on_consume)
        with close_many(components.client, components.conductor):
            t = threading_utils.daemon_thread(
                lambda: components.conductor.run(max_dispatches=5))
            t.start()
            lb, fd = pu.temporary_flow_detail(components.persistence)
            engines.save_factory_details(fd, test_factory,
                                         [False], {},
                                         backend=components.persistence)
            for _ in range(5):
                components.board.post('poke', lb,
                                      details={'flow_uuid': fd.uuid})
                self.assertTrue(consumed_event.wait(
                    test_utils.WAIT_TIMEOUT))
            components.board.post('poke', lb,
                                  details={'flow_uuid': fd.uuid})
            components.conductor.stop()
            self.assertTrue(components.conductor.wait(test_utils.WAIT_TIMEOUT))
            self.assertFalse(components.conductor.dispatching)

    def test_threaded_access_property(self):
        called = collections.deque()

        class A(object):
            @misc.cachedproperty
            def b(self):
                called.append(1)
                # NOTE(harlowja): wait for a little and give some time for
                # another thread to potentially also get in this method to
                # also create the same property...
                time.sleep(random.random() * 0.5)
                return 'b'

        a = A()
        threads = []
        try:
            for _i in range(0, 20):
                t = threading_utils.daemon_thread(lambda: a.b)
                threads.append(t)
            for t in threads:
                t.start()
        finally:
            while threads:
                t = threads.pop()
                t.join()

        self.assertEqual(1, len(called))
        self.assertEqual('b', a.b)

    def test_start_stop(self):
        events = collections.deque()

        def before_start(t):
            events.append('bs')

        def before_join(t):
            events.append('bj')
            self.death.set()

        def after_start(t):
            events.append('as')

        def after_join(t):
            events.append('aj')

        for _i in range(0, self.thread_count):
            self.bundle.bind(lambda: tu.daemon_thread(_spinner, self.death),
                             before_join=before_join,
                             after_join=after_join,
                             before_start=before_start,
                             after_start=after_start)
        self.assertEqual(self.thread_count, self.bundle.start())
        self.assertEqual(self.thread_count, len(self.bundle))
        self.assertEqual(self.thread_count, self.bundle.stop())
        for event in ['as', 'bs', 'bj', 'aj']:
            self.assertEqual(self.thread_count,
                             len([e for e in events if e == event]))
        self.assertEqual(0, self.bundle.stop())
        self.assertTrue(self.death.is_set())

    def test_stop(self):
        self.conn_inst_mock.drain_events.side_effect = socket.timeout

        # create proxy
        pr = self.proxy(reset_master_mock=True)

        # check that proxy is not running yes
        self.assertFalse(pr.is_running)

        # start proxy in separate thread
        t = threading_utils.daemon_thread(pr.start)
        t.start()

        # make sure proxy is started
        pr.wait()

        # check that proxy is running now
        self.assertTrue(pr.is_running)

        # stop proxy and wait for thread to finish
        pr.stop()

        # wait for thread to finish
        t.join()

        self.assertFalse(pr.is_running)

    def test_double_acquire_many(self):
        activated = collections.deque()
        acquires = collections.deque()
        n_lock = lock_utils.MultiLock((threading.RLock(), threading.RLock()))

        def critical_section():
            start = now()
            time.sleep(NAPPY_TIME)
            end = now()
            activated.append((start, end))

        def run():
            with n_lock as gotten:
                acquires.append(gotten)
                critical_section()
                with n_lock as gotten:
                    acquires.append(gotten)
                    critical_section()
                critical_section()

        threads = []
        for i in range(0, self.THREAD_COUNT):
            t = threading_utils.daemon_thread(run)
            threads.append(t)
            t.start()
        while threads:
            t = threads.pop()
            t.join()

        self.assertTrue(all(acquires))
        self.assertEqual(self.THREAD_COUNT * 2, len(acquires))
        self.assertEqual(self.THREAD_COUNT * 3, len(activated))
        for (start, end) in activated:
            self.assertEqual(1, _find_overlaps(activated, start, end))

    def test_multi_message(self):
        message_count = 30
        barrier = latch.Latch(message_count)
        countdown = lambda data, message: barrier.countdown()

        on_notify = mock.MagicMock()
        on_notify.side_effect = countdown

        on_response = mock.MagicMock()
        on_response.side_effect = countdown

        on_request = mock.MagicMock()
        on_request.side_effect = countdown

        handlers = {
            pr.NOTIFY: dispatcher.Handler(on_notify),
            pr.RESPONSE: dispatcher.Handler(on_response),
            pr.REQUEST: dispatcher.Handler(on_request),
        }
        p = proxy.Proxy(TEST_TOPIC, TEST_EXCHANGE, handlers,
                        transport='memory',
                        transport_options={
                            'polling_interval': POLLING_INTERVAL,
                        })

        t = threading_utils.daemon_thread(p.start)
        t.start()
        p.wait()

        for i in range(0, message_count):
            j = i % 3
            if j == 0:
                p.publish(pr.Notify(), TEST_TOPIC)
            elif j == 1:
                p.publish(pr.Response(pr.RUNNING), TEST_TOPIC)
            else:
                p.publish(pr.Request(test_utils.DummyTask("dummy_%s" % i),
                                     uuidutils.generate_uuid(),
                                     pr.EXECUTE, [], None), TEST_TOPIC)

        self.assertTrue(barrier.wait(test_utils.WAIT_TIMEOUT))
        self.assertEqual(0, barrier.needed)
        p.stop()
        t.join()

        self.assertTrue(on_notify.called)
        self.assertTrue(on_response.called)
        self.assertTrue(on_request.called)

        self.assertEqual(10, on_notify.call_count)
        self.assertEqual(10, on_response.call_count)
        self.assertEqual(10, on_request.call_count)

        call_count = sum([
            on_notify.call_count,
            on_response.call_count,
            on_request.call_count,
        ])
        self.assertEqual(message_count, call_count)

 def start(self):
     """Starts message processing thread."""
     if self._helper is not None:
         raise RuntimeError("Worker executor must be stopped before"
                            " it can be started")
     self._helper = tu.daemon_thread(self._proxy.start)
     self._helper.start()
     self._proxy.wait()

 def test_bundle_length(self):
     self.assertEqual(0, len(self.bundle))
     for i in range(0, self.thread_count):
         self.bundle.bind(lambda: tu.daemon_thread(_spinner, self.death))
         self.assertEqual(1, self.bundle.start())
         self.assertEqual(i + 1, len(self.bundle))
     self.death.set()
     self.assertEqual(self.thread_count, self.bundle.stop())
     self.assertEqual(self.thread_count, len(self.bundle))

 def test_alive_thread(self):
     death = tu.Event()
     t = tu.daemon_thread(_spinner, death)
     self.assertFalse(tu.is_alive(t))
     t.start()
     self.assertTrue(tu.is_alive(t))
     death.set()
     t.join()
     self.assertFalse(tu.is_alive(t))

 def start(self):
     if threading_utils.is_alive(self._worker):
         raise RuntimeError("Worker thread must be stopped via stop()"
                            " before starting/restarting")
     super(ParallelProcessTaskExecutor, self).start()
     self._dispatcher.setup()
     self._worker = threading_utils.daemon_thread(
         asyncore.loop, map=self._dispatcher.map,
         timeout=self._wait_timeout)
     self._worker.start()