Method test_notify

Lib/test/_test_multiprocessing.py:1615–1655

(self)

                pass

    def test_notify(self):
        cond = self.Condition()
        sleeping = self.Semaphore(0)
        woken = self.Semaphore(0)

        p = self.Process(target=self.f, args=(cond, sleeping, woken))
        p.daemon = True
        p.start()

        t = threading.Thread(target=self.f, args=(cond, sleeping, woken))
        t.daemon = True
        t.start()

        # wait for both children to start sleeping
        sleeping.acquire()
        sleeping.acquire()

        # check no process/thread has woken up
        self.assertReachesEventually(lambda: get_value(woken), 0)

        # wake up one process/thread
        cond.acquire()
        cond.notify()
        cond.release()

        # check one process/thread has woken up
        self.assertReachesEventually(lambda: get_value(woken), 1)

        # wake up another
        cond.acquire()
        cond.notify()
        cond.release()

        # check other has woken up
        self.assertReachesEventually(lambda: get_value(woken), 2)

        # check state is not mucked up
        self.check_invariant(cond)

        threading_helper.join_thread(t)
        join_process(p)

    def test_notify_all(self):
        cond = self.Condition()