#

# Unit tests for the multiprocessing package

#

import unittest

import queue as pyqueue

import contextlib

import time

import io

import itertools

import sys

import os

import gc

import errno

import signal

import array

import socket

import random

import logging

import struct

import operator

import weakref

import test.support

import test.support.script_helper

from test import support

# Skip tests if _multiprocessing wasn't built.

_multiprocessing = test.support.import_module('_multiprocessing')

# Skip tests if sem_open implementation is broken.

test.support.import_module('multiprocessing.synchronize')

# import threading after _multiprocessing to raise a more relevant error

# message: "No module named _multiprocessing". _multiprocessing is not compiled

# without thread support.

import threading

import multiprocessing.connection

import multiprocessing.dummy

import multiprocessing.heap

import multiprocessing.managers

import multiprocessing.pool

import multiprocessing.queues

from multiprocessing import util

try:

from multiprocessing import reduction

HAS_REDUCTION = reduction.HAVE_SEND_HANDLE

except ImportError:

HAS_REDUCTION = False

try:

from multiprocessing.sharedctypes import Value, copy

HAS_SHAREDCTYPES = True

except ImportError:

HAS_SHAREDCTYPES = False

try:

import msvcrt

except ImportError:

msvcrt = None

#

#

#

# Timeout to wait until a process completes

TIMEOUT = 60.0 # seconds

def latin(s):

return s.encode('latin')

def close_queue(queue):

if isinstance(queue, multiprocessing.queues.Queue):

queue.close()

queue.join_thread()

def join_process(process):

# Since multiprocessing.Process has the same API than threading.Thread

# (join() and is_alive(), the support function can be reused

support.join_thread(process, timeout=TIMEOUT)

#

# Constants

#

LOG_LEVEL = util.SUBWARNING

#LOG_LEVEL = logging.DEBUG

DELTA = 0.1

CHECK_TIMINGS = False # making true makes tests take a lot longer

# and can sometimes cause some non-serious

# failures because some calls block a bit

# longer than expected

if CHECK_TIMINGS:

TIMEOUT1, TIMEOUT2, TIMEOUT3 = 0.82, 0.35, 1.4

else:

TIMEOUT1, TIMEOUT2, TIMEOUT3 = 0.1, 0.1, 0.1

HAVE_GETVALUE = not getattr(_multiprocessing,

'HAVE_BROKEN_SEM_GETVALUE', False)

WIN32 = (sys.platform == "win32")

from multiprocessing.connection import wait

def wait_for_handle(handle, timeout):

if timeout is not None and timeout < 0.0:

timeout = None

return wait([handle], timeout)

try:

MAXFD = os.sysconf("SC_OPEN_MAX")

except:

MAXFD = 256

# To speed up tests when using the forkserver, we can preload these:

PRELOAD = ['__main__', 'test.test_multiprocessing_forkserver']

#

# Some tests require ctypes

#

try:

from ctypes import Structure, c_int, c_double, c_longlong

except ImportError:

Structure = object

c_int = c_double = c_longlong = None

def check_enough_semaphores():

"""Check that the system supports enough semaphores to run the test."""

# minimum number of semaphores available according to POSIX

nsems_min = 256

try:

nsems = os.sysconf("SC_SEM_NSEMS_MAX")

except (AttributeError, ValueError):

# sysconf not available or setting not available

return

if nsems == -1 or nsems >= nsems_min:

return

raise unittest.SkipTest("The OS doesn't support enough semaphores "

"to run the test (required: %d)." % nsems_min)

#

# Creates a wrapper for a function which records the time it takes to finish

#

class TimingWrapper(object):

def __init__(self, func):

self.func = func

self.elapsed = None

def __call__(self, *args, **kwds):

t = time.monotonic()

try:

return self.func(*args, **kwds)

finally:

self.elapsed = time.monotonic() - t

#

# Base class for test cases

#

class BaseTestCase(object):

ALLOWED_TYPES = ('processes', 'manager', 'threads')

def assertTimingAlmostEqual(self, a, b):

if CHECK_TIMINGS:

self.assertAlmostEqual(a, b, 1)

def assertReturnsIfImplemented(self, value, func, *args):

try:

res = func(*args)

except NotImplementedError:

pass

else:

return self.assertEqual(value, res)

# For the sanity of Windows users, rather than crashing or freezing in

# multiple ways.

def __reduce__(self, *args):

raise NotImplementedError("shouldn't try to pickle a test case")

__reduce_ex__ = __reduce__

#

# Return the value of a semaphore

#

def get_value(self):

try:

return self.get_value()

except AttributeError:

try:

return self._Semaphore__value

except AttributeError:

try:

return self._value

except AttributeError:

raise NotImplementedError

#

# Testcases

#

class DummyCallable:

def __call__(self, q, c):

assert isinstance(c, DummyCallable)

q.put(5)

class _TestProcess(BaseTestCase):

ALLOWED_TYPES = ('processes', 'threads')

def test_current(self):

if self.TYPE == 'threads':

self.skipTest('test not appropriate for {}'.format(self.TYPE))

current = self.current_process()

authkey = current.authkey

self.assertTrue(current.is_alive())

self.assertTrue(not current.daemon)

self.assertIsInstance(authkey, bytes)

self.assertTrue(len(authkey) > 0)

self.assertEqual(current.ident, os.getpid())

self.assertEqual(current.exitcode, None)

def test_daemon_argument(self):

if self.TYPE == "threads":

self.skipTest('test not appropriate for {}'.format(self.TYPE))

# By default uses the current process's daemon flag.

proc0 = self.Process(target=self._test)

self.assertEqual(proc0.daemon, self.current_process().daemon)

proc1 = self.Process(target=self._test, daemon=True)

self.assertTrue(proc1.daemon)

proc2 = self.Process(target=self._test, daemon=False)

self.assertFalse(proc2.daemon)

@classmethod

def _test(cls, q, *args, **kwds):

current = cls.current_process()

q.put(args)

q.put(kwds)

q.put(current.name)

if cls.TYPE != 'threads':

q.put(bytes(current.authkey))

q.put(current.pid)

def test_process(self):

q = self.Queue(1)

e = self.Event()

args = (q, 1, 2)

kwargs = {'hello':23, 'bye':2.54}

name = 'SomeProcess'

p = self.Process(

target=self._test, args=args, kwargs=kwargs, name=name

)

p.daemon = True

current = self.current_process()

if self.TYPE != 'threads':

self.assertEqual(p.authkey, current.authkey)

self.assertEqual(p.is_alive(), False)

self.assertEqual(p.daemon, True)

self.assertNotIn(p, self.active_children())

self.assertTrue(type(self.active_children()) is list)

self.assertEqual(p.exitcode, None)

p.start()

self.assertEqual(p.exitcode, None)

self.assertEqual(p.is_alive(), True)

self.assertIn(p, self.active_children())

self.assertEqual(q.get(), args[1:])

self.assertEqual(q.get(), kwargs)

self.assertEqual(q.get(), p.name)

if self.TYPE != 'threads':

self.assertEqual(q.get(), current.authkey)

self.assertEqual(q.get(), p.pid)

p.join()

self.assertEqual(p.exitcode, 0)

self.assertEqual(p.is_alive(), False)

self.assertNotIn(p, self.active_children())

close_queue(q)

@classmethod

def _sleep_some(cls):

time.sleep(100)

@classmethod

def _test_sleep(cls, delay):

time.sleep(delay)

def _kill_process(self, meth):

if self.TYPE == 'threads':

self.skipTest('test not appropriate for {}'.format(self.TYPE))

p = self.Process(target=self._sleep_some)

p.daemon = True

p.start()

self.assertEqual(p.is_alive(), True)

self.assertIn(p, self.active_children())

self.assertEqual(p.exitcode, None)

join = TimingWrapper(p.join)

self.assertEqual(join(0), None)

self.assertTimingAlmostEqual(join.elapsed, 0.0)

self.assertEqual(p.is_alive(), True)

self.assertEqual(join(-1), None)

self.assertTimingAlmostEqual(join.elapsed, 0.0)

self.assertEqual(p.is_alive(), True)

# XXX maybe terminating too soon causes the problems on Gentoo...

time.sleep(1)

meth(p)

if hasattr(signal, 'alarm'):

# On the Gentoo buildbot waitpid() often seems to block forever.

# We use alarm() to interrupt it if it blocks for too long.

def handler(*args):

raise RuntimeError('join took too long: %s' % p)

old_handler = signal.signal(signal.SIGALRM, handler)

try:

signal.alarm(10)

self.assertEqual(join(), None)

finally:

signal.alarm(0)

signal.signal(signal.SIGALRM, old_handler)

else:

self.assertEqual(join(), None)

self.assertTimingAlmostEqual(join.elapsed, 0.0)

self.assertEqual(p.is_alive(), False)

self.assertNotIn(p, self.active_children())

p.join()

return p.exitcode

def test_terminate(self):

exitcode = self._kill_process(multiprocessing.Process.terminate)

if os.name != 'nt':

self.assertEqual(exitcode, -signal.SIGTERM)

def test_kill(self):

exitcode = self._kill_process(multiprocessing.Process.kill)

if os.name != 'nt':

self.assertEqual(exitcode, -signal.SIGKILL)

def test_cpu_count(self):

try:

cpus = multiprocessing.cpu_count()

except NotImplementedError:

cpus = 1

self.assertTrue(type(cpus) is int)

self.assertTrue(cpus >= 1)

def test_active_children(self):

self.assertEqual(type(self.active_children()), list)

p = self.Process(target=time.sleep, args=(DELTA,))

self.assertNotIn(p, self.active_children())

p.daemon = True

p.start()

self.assertIn(p, self.active_children())

p.join()

self.assertNotIn(p, self.active_children())

@classmethod

def _test_recursion(cls, wconn, id):

wconn.send(id)

if len(id) < 2:

for i in range(2):

p = cls.Process(

target=cls._test_recursion, args=(wconn, id+[i])

)

p.start()

p.join()

def test_recursion(self):

rconn, wconn = self.Pipe(duplex=False)

self._test_recursion(wconn, [])

time.sleep(DELTA)

result = []

while rconn.poll():

result.append(rconn.recv())

expected = [

[],

[0],

[0, 0],

[0, 1],

[1],

[1, 0],

[1, 1]

]

self.assertEqual(result, expected)

@classmethod

def _test_sentinel(cls, event):

event.wait(10.0)

def test_sentinel(self):

if self.TYPE == "threads":

self.skipTest('test not appropriate for {}'.format(self.TYPE))

event = self.Event()

p = self.Process(target=self._test_sentinel, args=(event,))

with self.assertRaises(ValueError):

p.sentinel

p.start()

self.addCleanup(p.join)

sentinel = p.sentinel

self.assertIsInstance(sentinel, int)

self.assertFalse(wait_for_handle(sentinel, timeout=0.0))

event.set()

p.join()

self.assertTrue(wait_for_handle(sentinel, timeout=1))

@classmethod

def _test_close(cls, rc=0, q=None):

if q is not None:

q.get()

sys.exit(rc)

def test_close(self):

if self.TYPE == "threads":

self.skipTest('test not appropriate for {}'.format(self.TYPE))

q = self.Queue()

p = self.Process(target=self._test_close, kwargs={'q': q})

p.daemon = True

p.start()

self.assertEqual(p.is_alive(), True)

# Child is still alive, cannot close

with self.assertRaises(ValueError):

p.close()

q.put(None)

p.join()

self.assertEqual(p.is_alive(), False)

self.assertEqual(p.exitcode, 0)

p.close()

with self.assertRaises(ValueError):

p.is_alive()

with self.assertRaises(ValueError):

p.join()

with self.assertRaises(ValueError):

p.terminate()

p.close()

wr = weakref.ref(p)

del p

gc.collect()

self.assertIs(wr(), None)

close_queue(q)

def test_many_processes(self):

if self.TYPE == 'threads':

self.skipTest('test not appropriate for {}'.format(self.TYPE))

sm = multiprocessing.get_start_method()

N = 5 if sm == 'spawn' else 100

# Try to overwhelm the forkserver loop with events

procs = [self.Process(target=self._test_sleep, args=(0.01,))

for i in range(N)]

for p in procs:

p.start()

for p in procs:

join_process(p)

for p in procs:

self.assertEqual(p.exitcode, 0)

procs = [self.Process(target=self._sleep_some)

for i in range(N)]

for p in procs:

p.start()

time.sleep(0.001) # let the children start...

for p in procs:

p.terminate()

for p in procs:

join_process(p)

if os.name != 'nt':

exitcodes = [-signal.SIGTERM]

if sys.platform == 'darwin':

# bpo-31510: On macOS, killing a freshly started process with

# SIGTERM sometimes kills the process with SIGKILL.

exitcodes.append(-signal.SIGKILL)

for p in procs:

self.assertIn(p.exitcode, exitcodes)

def test_lose_target_ref(self):

c = DummyCallable()

wr = weakref.ref(c)

q = self.Queue()

p = self.Process(target=c, args=(q, c))

del c

p.start()

p.join()

self.assertIs(wr(), None)

self.assertEqual(q.get(), 5)

close_queue(q)

@classmethod

def _test_child_fd_inflation(self, evt, q):

q.put(test.support.fd_count())

evt.wait()

def test_child_fd_inflation(self):

# Number of fds in child processes should not grow with the

# number of running children.

if self.TYPE == 'threads':

self.skipTest('test not appropriate for {}'.format(self.TYPE))

sm = multiprocessing.get_start_method()

if sm == 'fork':

# The fork method by design inherits all fds from the parent,

# trying to go against it is a lost battle

self.skipTest('test not appropriate for {}'.format(sm))

N = 5

evt = self.Event()

q = self.Queue()

procs = [self.Process(target=self._test_child_fd_inflation, args=(evt, q))

for i in range(N)]

for p in procs:

p.start()

try:

fd_counts = [q.get() for i in range(N)]

self.assertEqual(len(set(fd_counts)), 1, fd_counts)

finally:

evt.set()

for p in procs:

p.join()

close_queue(q)

@classmethod

def _test_wait_for_threads(self, evt):

def func1():

time.sleep(0.5)

evt.set()

def func2():

time.sleep(20)

evt.clear()

threading.Thread(target=func1).start()

threading.Thread(target=func2, daemon=True).start()

def test_wait_for_threads(self):

# A child process should wait for non-daemonic threads to end

# before exiting

if self.TYPE == 'threads':

self.skipTest('test not appropriate for {}'.format(self.TYPE))

evt = self.Event()

proc = self.Process(target=self._test_wait_for_threads, args=(evt,))

proc.start()

proc.join()

self.assertTrue(evt.is_set())

@classmethod

def _test_error_on_stdio_flush(self, evt, break_std_streams={}):

for stream_name, action in break_std_streams.items():

if action == 'close':

stream = io.StringIO()

stream.close()

else:

assert action == 'remove'

stream = None

setattr(sys, stream_name, None)

evt.set()

def test_error_on_stdio_flush_1(self):

# Check that Process works with broken standard streams

streams = [io.StringIO(), None]

streams[0].close()

for stream_name in ('stdout', 'stderr'):

for stream in streams:

old_stream = getattr(sys, stream_name)

setattr(sys, stream_name, stream)

try:

evt = self.Event()

proc = self.Process(target=self._test_error_on_stdio_flush,

args=(evt,))

proc.start()

proc.join()

self.assertTrue(evt.is_set())

self.assertEqual(proc.exitcode, 0)

finally:

setattr(sys, stream_name, old_stream)

def test_error_on_stdio_flush_2(self):

# Same as test_error_on_stdio_flush_1(), but standard streams are

# broken by the child process

for stream_name in ('stdout', 'stderr'):

for action in ('close', 'remove'):

old_stream = getattr(sys, stream_name)

try:

evt = self.Event()

proc = self.Process(target=self._test_error_on_stdio_flush,

args=(evt, {stream_name: action}))

proc.start()

proc.join()

self.assertTrue(evt.is_set())

self.assertEqual(proc.exitcode, 0)

finally:

setattr(sys, stream_name, old_stream)

@classmethod

def _sleep_and_set_event(self, evt, delay=0.0):

time.sleep(delay)

evt.set()

def check_forkserver_death(self, signum):

# bpo-31308: if the forkserver process has died, we should still

# be able to create and run new Process instances (the forkserver

# is implicitly restarted).

if self.TYPE == 'threads':

self.skipTest('test not appropriate for {}'.format(self.TYPE))

sm = multiprocessing.get_start_method()

if sm != 'forkserver':

# The fork method by design inherits all fds from the parent,

# trying to go against it is a lost battle

self.skipTest('test not appropriate for {}'.format(sm))

from multiprocessing.forkserver import _forkserver

_forkserver.ensure_running()

# First process sleeps 500 ms

delay = 0.5

evt = self.Event()

proc = self.Process(target=self._sleep_and_set_event, args=(evt, delay))

proc.start()

pid = _forkserver._forkserver_pid

os.kill(pid, signum)

# give time to the fork server to die and time to proc to complete

time.sleep(delay * 2.0)

evt2 = self.Event()

proc2 = self.Process(target=self._sleep_and_set_event, args=(evt2,))

proc2.start()

proc2.join()

self.assertTrue(evt2.is_set())

self.assertEqual(proc2.exitcode, 0)

proc.join()

self.assertTrue(evt.is_set())

self.assertIn(proc.exitcode, (0, 255))

def test_forkserver_sigint(self):

# Catchable signal

self.check_forkserver_death(signal.SIGINT)

def test_forkserver_sigkill(self):

# Uncatchable signal

if os.name != 'nt':

self.check_forkserver_death(signal.SIGKILL)

#

#

#

class _UpperCaser(multiprocessing.Process):

def __init__(self):

multiprocessing.Process.__init__(self)

self.child_conn, self.parent_conn = multiprocessing.Pipe()

def run(self):

self.parent_conn.close()

for s in iter(self.child_conn.recv, None):

self.child_conn.send(s.upper())

self.child_conn.close()

def submit(self, s):

assert type(s) is str

self.parent_conn.send(s)

return self.parent_conn.recv()

def stop(self):

self.parent_conn.send(None)

self.parent_conn.close()

self.child_conn.close()

class _TestSubclassingProcess(BaseTestCase):

ALLOWED_TYPES = ('processes',)

def test_subclassing(self):

uppercaser = _UpperCaser()

uppercaser.daemon = True

uppercaser.start()

self.assertEqual(uppercaser.submit('hello'), 'HELLO')

self.assertEqual(uppercaser.submit('world'), 'WORLD')

uppercaser.stop()

uppercaser.join()

def test_stderr_flush(self):

# sys.stderr is flushed at process shutdown (issue #13812)

if self.TYPE == "threads":

self.skipTest('test not appropriate for {}'.format(self.TYPE))

testfn = test.support.TESTFN

self.addCleanup(test.support.unlink, testfn)

proc = self.Process(target=self._test_stderr_flush, args=(testfn,))

proc.start()

proc.join()

with open(testfn, 'r') as f:

err = f.read()

# The whole traceback was printed

self.assertIn("ZeroDivisionError", err)

self.assertIn("test_multiprocessing.py", err)

self.assertIn("1/0 # MARKER", err)

@classmethod

def _test_stderr_flush(cls, testfn):

fd = os.open(testfn, os.O_WRONLY | os.O_CREAT | os.O_EXCL)

sys.stderr = open(fd, 'w', closefd=False)

1/0 # MARKER

@classmethod

def _test_sys_exit(cls, reason, testfn):

fd = os.open(testfn, os.O_WRONLY | os.O_CREAT | os.O_EXCL)

sys.stderr = open(fd, 'w', closefd=False)

sys.exit(reason)

def test_sys_exit(self):

# See Issue 13854

if self.TYPE == 'threads':

self.skipTest('test not appropriate for {}'.format(self.TYPE))

testfn = test.support.TESTFN

self.addCleanup(test.support.unlink, testfn)

for reason in (

[1, 2, 3],

'ignore this',

):

p = self.Process(target=self._test_sys_exit, args=(reason, testfn))

p.daemon = True

p.start()

join_process(p)

self.assertEqual(p.exitcode, 1)

with open(testfn, 'r') as f:

content = f.read()

self.assertEqual(content.rstrip(), str(reason))

os.unlink(testfn)

for reason in (True, False, 8):

p = self.Process(target=sys.exit, args=(reason,))

p.daemon = True

p.start()

join_process(p)

self.assertEqual(p.exitcode, reason)

#

#

#

def queue_empty(q):

if hasattr(q, 'empty'):

return q.empty()

else:

return q.qsize() == 0

def queue_full(q, maxsize):

if hasattr(q, 'full'):

return q.full()

else:

return q.qsize() == maxsize

class _TestQueue(BaseTestCase):

@classmethod

def _test_put(cls, queue, child_can_start, parent_can_continue):

child_can_start.wait()

for i in range(6):

queue.get()

parent_can_continue.set()

def test_put(self):

MAXSIZE = 6

queue = self.Queue(maxsize=MAXSIZE)

child_can_start = self.Event()

parent_can_continue = self.Event()

proc = self.Process(

target=self._test_put,

args=(queue, child_can_start, parent_can_continue)

)

proc.daemon = True

proc.start()

self.assertEqual(queue_empty(queue), True)

self.assertEqual(queue_full(queue, MAXSIZE), False)

queue.put(1)

queue.put(2, True)

queue.put(3, True, None)

queue.put(4, False)

queue.put(5, False, None)

queue.put_nowait(6)

# the values may be in buffer but not yet in pipe so sleep a bit

time.sleep(DELTA)

self.assertEqual(queue_empty(queue), False)

self.assertEqual(queue_full(queue, MAXSIZE), True)

put = TimingWrapper(queue.put)

put_nowait = TimingWrapper(queue.put_nowait)

self.assertRaises(pyqueue.Full, put, 7, False)

self.assertTimingAlmostEqual(put.elapsed, 0)

self.assertRaises(pyqueue.Full, put, 7, False, None)

self.assertTimingAlmostEqual(put.elapsed, 0)

self.assertRaises(pyqueue.Full, put_nowait, 7)

self.assertTimingAlmostEqual(put_nowait.elapsed, 0)

self.assertRaises(pyqueue.Full, put, 7, True, TIMEOUT1)

self.assertTimingAlmostEqual(put.elapsed, TIMEOUT1)

self.assertRaises(pyqueue.Full, put, 7, False, TIMEOUT2)

self.assertTimingAlmostEqual(put.elapsed, 0)

self.assertRaises(pyqueue.Full, put, 7, True, timeout=TIMEOUT3)

self.assertTimingAlmostEqual(put.elapsed, TIMEOUT3)

child_can_start.set()

parent_can_continue.wait()

self.assertEqual(queue_empty(queue), True)

self.assertEqual(queue_full(queue, MAXSIZE), False)

proc.join()

close_queue(queue)

@classmethod

def _test_get(cls, queue, child_can_start, parent_can_continue):

child_can_start.wait()

#queue.put(1)

queue.put(2)

queue.put(3)

queue.put(4)

queue.put(5)

parent_can_continue.set()

def test_get(self):

queue = self.Queue()

child_can_start = self.Event()

parent_can_continue = self.Event()

proc = self.Process(

target=self._test_get,

args=(queue, child_can_start, parent_can_continue)

)

proc.daemon = True

proc.start()

self.assertEqual(queue_empty(queue), True)

child_can_start.set()

parent_can_continue.wait()

time.sleep(DELTA)

self.assertEqual(queue_empty(queue), False)

# Hangs unexpectedly, remove for now

#self.assertEqual(queue.get(), 1)

self.assertEqual(queue.get(True, None), 2)

self.assertEqual(queue.get(True), 3)

self.assertEqual(queue.get(timeout=1), 4)

self.assertEqual(queue.get_nowait(), 5)

self.assertEqual(queue_empty(queue), True)

get = TimingWrapper(queue.get)

get_nowait = TimingWrapper(queue.get_nowait)

self.assertRaises(pyqueue.Empty, get, False)

self.assertTimingAlmostEqual(get.elapsed, 0)

self.assertRaises(pyqueue.Empty, get, False, None)

self.assertTimingAlmostEqual(get.elapsed, 0)

self.assertRaises(pyqueue.Empty, get_nowait)

self.assertTimingAlmostEqual(get_nowait.elapsed, 0)

self.assertRaises(pyqueue.Empty, get, True, TIMEOUT1)

self.assertTimingAlmostEqual(get.elapsed, TIMEOUT1)

self.assertRaises(pyqueue.Empty, get, False, TIMEOUT2)

self.assertTimingAlmostEqual(get.elapsed, 0)

self.assertRaises(pyqueue.Empty, get, timeout=TIMEOUT3)

self.assertTimingAlmostEqual(get.elapsed, TIMEOUT3)

proc.join()

close_queue(queue)

@classmethod

def _test_fork(cls, queue):

for i in range(10, 20):

queue.put(i)

# note that at this point the items may only be buffered, so the

# process cannot shutdown until the feeder thread has finished

# pushing items onto the pipe.

def test_fork(self):

# Old versions of Queue would fail to create a new feeder

# thread for a forked process if the original process had its

# own feeder thread. This test checks that this no longer

# happens.

queue = self.Queue()

# put items on queue so that main process starts a feeder thread

for i in range(10):

queue.put(i)

# wait to make sure thread starts before we fork a new process

time.sleep(DELTA)

# fork process

p = self.Process(target=self._test_fork, args=(queue,))

p.daemon = True

p.start()

# check that all expected items are in the queue

for i in range(20):

self.assertEqual(queue.get(), i)

self.assertRaises(pyqueue.Empty, queue.get, False)

p.join()

close_queue(queue)

def test_qsize(self):

q = self.Queue()

try:

self.assertEqual(q.qsize(), 0)

except NotImplementedError:

self.skipTest('qsize method not implemented')

q.put(1)

self.assertEqual(q.qsize(), 1)

q.put(5)

self.assertEqual(q.qsize(), 2)

q.get()

self.assertEqual(q.qsize(), 1)

q.get()

self.assertEqual(q.qsize(), 0)

close_queue(q)

@classmethod

def _test_task_done(cls, q):

for obj in iter(q.get, None):

time.sleep(DELTA)

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: