Method test_join_from_self

Lib/test/test_thread.py:215–242

(self)

            handle_destroyed.release()

    def test_join_from_self(self):
        errors = []
        handles = []
        start_joinable_thread_returned = thread.allocate_lock()
        start_joinable_thread_returned.acquire()
        task_tried_to_join = thread.allocate_lock()
        task_tried_to_join.acquire()

        def task():
            start_joinable_thread_returned.acquire()
            try:
                handles[0].join()
            except Exception as e:
                errors.append(e)
            finally:
                task_tried_to_join.release()

        with threading_helper.wait_threads_exit():
            handle = thread.start_joinable_thread(task)
            handles.append(handle)
            start_joinable_thread_returned.release()
            # Can still join after joining failed in other thread
            task_tried_to_join.acquire()
            handle.join()

        assert len(errors) == 1
        with self.assertRaisesRegex(RuntimeError, "Cannot join current thread"):
            raise errors[0]

    def test_join_then_self_join(self):
        # make sure we can't deadlock in the following scenario with