Method test_task_done

Lib/test/_test_multiprocessing.py:1231–1251

(self)

            q.task_done()

    def test_task_done(self):
        queue = self.JoinableQueue()

        workers = [self.Process(target=self._test_task_done, args=(queue,))
                   for i in range(4)]

        for p in workers:
            p.daemon = True
            p.start()

        for i in range(10):
            queue.put(i)

        queue.join()

        for p in workers:
            queue.put(None)

        for p in workers:
            p.join()
        close_queue(queue)

    def test_no_import_lock_contention(self):
        with os_helper.temp_cwd():