_test_multiprocessing.py

#
# Unit tests for the multiprocessing package
#

importunittest
importqueueaspyqueue
importcontextlib
importtime
importio
importitertools
importsys
importos
importgc
importerrno
importsignal
importarray
importsocket
importrandom
importlogging
importstruct
importoperator
importweakref
importtest.support
importtest.support.script_helper
fromtestimportsupport


# 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.
importthreading

importmultiprocessing.connection
importmultiprocessing.dummy
importmultiprocessing.heap
importmultiprocessing.managers
importmultiprocessing.pool
importmultiprocessing.queues

frommultiprocessingimportutil

try:
frommultiprocessingimportreduction
HAS_REDUCTION=reduction.HAVE_SEND_HANDLE
exceptImportError:
HAS_REDUCTION=False

try:
frommultiprocessing.sharedctypesimportValue, copy
HAS_SHAREDCTYPES=True
exceptImportError:
HAS_SHAREDCTYPES=False

try:
importmsvcrt
exceptImportError:
msvcrt=None

#
#
#

# Timeout to wait until a process completes
TIMEOUT=60.0# seconds

deflatin(s):
returns.encode('latin')


defclose_queue(queue):
ifisinstance(queue, multiprocessing.queues.Queue):
queue.close()
queue.join_thread()


defjoin_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
ifCHECK_TIMINGS:
TIMEOUT1, TIMEOUT2, TIMEOUT3=0.82, 0.35, 1.4
else:
TIMEOUT1, TIMEOUT2, TIMEOUT3=0.1, 0.1, 0.1

HAVE_GETVALUE=notgetattr(_multiprocessing,
'HAVE_BROKEN_SEM_GETVALUE', False)

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

frommultiprocessing.connectionimportwait

defwait_for_handle(handle, timeout):
iftimeoutisnotNoneandtimeout<0.0:
timeout=None
returnwait([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:
fromctypesimportStructure, c_int, c_double, c_longlong
exceptImportError:
Structure=object
c_int=c_double=c_longlong=None


defcheck_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
ifnsems==-1ornsems>=nsems_min:
return
raiseunittest.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
#

classTimingWrapper(object):

def__init__(self, func):
self.func=func
self.elapsed=None

def__call__(self, *args, **kwds):
t=time.monotonic()
try:
returnself.func(*args, **kwds)
finally:
self.elapsed=time.monotonic() -t

#
# Base class for test cases
#

classBaseTestCase(object):

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

defassertTimingAlmostEqual(self, a, b):
ifCHECK_TIMINGS:
self.assertAlmostEqual(a, b, 1)

defassertReturnsIfImplemented(self, value, func, *args):
try:
res=func(*args)
exceptNotImplementedError:
pass
else:
returnself.assertEqual(value, res)

# For the sanity of Windows users, rather than crashing or freezing in
# multiple ways.
def__reduce__(self, *args):
raiseNotImplementedError("shouldn't try to pickle a test case")

__reduce_ex__=__reduce__

#
# Return the value of a semaphore
#

defget_value(self):
try:
returnself.get_value()
exceptAttributeError:
try:
returnself._Semaphore__value
exceptAttributeError:
try:
returnself._value
exceptAttributeError:
raiseNotImplementedError

#
# Testcases
#

classDummyCallable:
def__call__(self, q, c):
assertisinstance(c, DummyCallable)
q.put(5)


class_TestProcess(BaseTestCase):

ALLOWED_TYPES= ('processes', 'threads')

deftest_current(self):
ifself.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(notcurrent.daemon)
self.assertIsInstance(authkey, bytes)
self.assertTrue(len(authkey) >0)
self.assertEqual(current.ident, os.getpid())
self.assertEqual(current.exitcode, None)

deftest_daemon_argument(self):
ifself.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)
ifcls.TYPE!='threads':
q.put(bytes(current.authkey))
q.put(current.pid)

deftest_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()

ifself.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()) islist)
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)
ifself.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):
ifself.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)

ifhasattr(signal, 'alarm'):
# On the Gentoo buildbot waitpid() often seems to block forever.
# We use alarm() to interrupt it if it blocks for too long.
defhandler(*args):
raiseRuntimeError('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()

returnp.exitcode

deftest_terminate(self):
exitcode=self._kill_process(multiprocessing.Process.terminate)
ifos.name!='nt':
self.assertEqual(exitcode, -signal.SIGTERM)

deftest_kill(self):
exitcode=self._kill_process(multiprocessing.Process.kill)
ifos.name!='nt':
self.assertEqual(exitcode, -signal.SIGKILL)

deftest_cpu_count(self):
try:
cpus=multiprocessing.cpu_count()
exceptNotImplementedError:
cpus=1
self.assertTrue(type(cpus) isint)
self.assertTrue(cpus>=1)

deftest_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)
iflen(id) <2:
foriinrange(2):
p=cls.Process(
target=cls._test_recursion, args=(wconn, id+[i])
 )
p.start()
p.join()

deftest_recursion(self):
rconn, wconn=self.Pipe(duplex=False)
self._test_recursion(wconn, [])

time.sleep(DELTA)
result= []
whilerconn.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)

deftest_sentinel(self):
ifself.TYPE=="threads":
self.skipTest('test not appropriate for {}'.format(self.TYPE))
event=self.Event()
p=self.Process(target=self._test_sentinel, args=(event,))
withself.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):
ifqisnotNone:
q.get()
sys.exit(rc)

deftest_close(self):
ifself.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
withself.assertRaises(ValueError):
p.close()

q.put(None)
p.join()
self.assertEqual(p.is_alive(), False)
self.assertEqual(p.exitcode, 0)
p.close()
withself.assertRaises(ValueError):
p.is_alive()
withself.assertRaises(ValueError):
p.join()
withself.assertRaises(ValueError):
p.terminate()
p.close()

wr=weakref.ref(p)
delp
gc.collect()
self.assertIs(wr(), None)

close_queue(q)

deftest_many_processes(self):
ifself.TYPE=='threads':
self.skipTest('test not appropriate for {}'.format(self.TYPE))

sm=multiprocessing.get_start_method()
N=5ifsm=='spawn'else100

# Try to overwhelm the forkserver loop with events
procs= [self.Process(target=self._test_sleep, args=(0.01,))
foriinrange(N)]
forpinprocs:
p.start()
forpinprocs:
join_process(p)
forpinprocs:
self.assertEqual(p.exitcode, 0)

procs= [self.Process(target=self._sleep_some)
foriinrange(N)]
forpinprocs:
p.start()
time.sleep(0.001) # let the children start...
forpinprocs:
p.terminate()
forpinprocs:
join_process(p)
ifos.name!='nt':
exitcodes= [-signal.SIGTERM]
ifsys.platform=='darwin':
# bpo-31510: On macOS, killing a freshly started process with
# SIGTERM sometimes kills the process with SIGKILL.
exitcodes.append(-signal.SIGKILL)
forpinprocs:
self.assertIn(p.exitcode, exitcodes)

deftest_lose_target_ref(self):
c=DummyCallable()
wr=weakref.ref(c)
q=self.Queue()
p=self.Process(target=c, args=(q, c))
delc
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()

deftest_child_fd_inflation(self):
# Number of fds in child processes should not grow with the
# number of running children.
ifself.TYPE=='threads':
self.skipTest('test not appropriate for {}'.format(self.TYPE))

sm=multiprocessing.get_start_method()
ifsm=='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))
foriinrange(N)]
forpinprocs:
p.start()

try:
fd_counts= [q.get() foriinrange(N)]
self.assertEqual(len(set(fd_counts)), 1, fd_counts)

finally:
evt.set()
forpinprocs:
p.join()
close_queue(q)

@classmethod
def_test_wait_for_threads(self, evt):
deffunc1():
time.sleep(0.5)
evt.set()

deffunc2():
time.sleep(20)
evt.clear()

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

deftest_wait_for_threads(self):
# A child process should wait for non-daemonic threads to end
# before exiting
ifself.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={}):
forstream_name, actioninbreak_std_streams.items():
ifaction=='close':
stream=io.StringIO()
stream.close()
else:
assertaction=='remove'
stream=None
setattr(sys, stream_name, None)
evt.set()

deftest_error_on_stdio_flush_1(self):
# Check that Process works with broken standard streams
streams= [io.StringIO(), None]
streams[0].close()
forstream_namein ('stdout', 'stderr'):
forstreaminstreams:
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)

deftest_error_on_stdio_flush_2(self):
# Same as test_error_on_stdio_flush_1(), but standard streams are
# broken by the child process
forstream_namein ('stdout', 'stderr'):
foractionin ('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()

defcheck_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).
ifself.TYPE=='threads':
self.skipTest('test not appropriate for {}'.format(self.TYPE))
sm=multiprocessing.get_start_method()
ifsm!='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))

frommultiprocessing.forkserverimport_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))

deftest_forkserver_sigint(self):
# Catchable signal
self.check_forkserver_death(signal.SIGINT)

deftest_forkserver_sigkill(self):
# Uncatchable signal
ifos.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()

defrun(self):
self.parent_conn.close()
forsiniter(self.child_conn.recv, None):
self.child_conn.send(s.upper())
self.child_conn.close()

defsubmit(self, s):
asserttype(s) isstr
self.parent_conn.send(s)
returnself.parent_conn.recv()

defstop(self):
self.parent_conn.send(None)
self.parent_conn.close()
self.child_conn.close()

class_TestSubclassingProcess(BaseTestCase):

ALLOWED_TYPES= ('processes',)

deftest_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()

deftest_stderr_flush(self):
# sys.stderr is flushed at process shutdown (issue #13812)
ifself.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()
withopen(testfn, 'r') asf:
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)

deftest_sys_exit(self):
# See Issue 13854
ifself.TYPE=='threads':
self.skipTest('test not appropriate for {}'.format(self.TYPE))

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

forreasonin (
 [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)

withopen(testfn, 'r') asf:
content=f.read()
self.assertEqual(content.rstrip(), str(reason))

os.unlink(testfn)

forreasonin (True, False, 8):
p=self.Process(target=sys.exit, args=(reason,))
p.daemon=True
p.start()
join_process(p)
self.assertEqual(p.exitcode, reason)

#
#
#

defqueue_empty(q):
ifhasattr(q, 'empty'):
returnq.empty()
else:
returnq.qsize() ==0

defqueue_full(q, maxsize):
ifhasattr(q, 'full'):
returnq.full()
else:
returnq.qsize() ==maxsize


class_TestQueue(BaseTestCase):


@classmethod
def_test_put(cls, queue, child_can_start, parent_can_continue):
child_can_start.wait()
foriinrange(6):
queue.get()
parent_can_continue.set()

deftest_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()

deftest_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):
foriinrange(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.

deftest_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
foriinrange(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
foriinrange(20):
self.assertEqual(queue.get(), i)
self.assertRaises(pyqueue.Empty, queue.get, False)

p.join()
close_queue(queue)

deftest_qsize(self):
q=self.Queue()
try:
self.assertEqual(q.qsize(), 0)
exceptNotImplementedError:
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):
forobjiniter(q.get, None):
time.sleep(DELTA)
q.task_done()

deftest_task_done(self):
queue=self.JoinableQueue()

workers= [self.Process(target=self._test_task_done, args=(queue,))
foriinrange(4)]

forpinworkers: