"""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 "

def __init__(self, func):

self.func = func

self.elapsed = None

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

t = time.time()

try:

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

finally:

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")

try:

return self.get_value()

except AttributeError:

try:

return self._Semaphore__value

except AttributeError:

try:

return self._value

except AttributeError:

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

@classmethod

def _test_terminate(cls):

time.sleep(100)

def test_terminate(self):

if self.TYPE == 'threads':

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

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

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)

p.terminate()

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

# XXX sometimes get p.exitcode == 0 on Windows ...

#self.assertEqual(p.exitcode, -signal.SIGTERM)

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

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

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

p.join(5)

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

p.join(5)

if hasattr(q, 'empty'):

return q.empty()

else:

if hasattr(q, 'full'):

return q.full()

else:

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

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

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

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)

@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:

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

def test_no_import_lock_contention(self):

with test.support.temp_cwd():

module_name = 'imported_by_an_imported_module'

with open(module_name + '.py', 'w') as f:

f.write("""if 1:

""")

with test.support.DirsOnSysPath(os.getcwd()):

try:

__import__(module_name)

except pyqueue.Empty:

self.fail("Probable regression on import lock contention;"

" see Issue #22853")

def test_timeout(self):

q = multiprocessing.Queue()

start = time.time()

self.assertRaises(pyqueue.Empty, q.get, True, 0.200)

delta = time.time() - start

# Tolerate a delta of 30 ms because of the bad clock resolution on

# Windows (usually 15.6 ms)

def test_lock(self):

lock = self.Lock()

self.assertEqual(lock.acquire(), True)

self.assertEqual(lock.acquire(False), False)

self.assertEqual(lock.release(), None)

self.assertRaises((ValueError, threading.ThreadError), lock.release)

def test_rlock(self):

lock = self.RLock()

self.assertEqual(lock.acquire(), True)

self.assertEqual(lock.acquire(), True)

self.assertEqual(lock.acquire(), True)

self.assertEqual(lock.release(), None)

self.assertEqual(lock.release(), None)

self.assertEqual(lock.release(), None)

self.assertRaises((AssertionError, RuntimeError), lock.release)

def test_lock_context(self):

with self.Lock():

def _test_semaphore(self, sem):

self.assertReturnsIfImplemented(2, get_value, sem)

self.assertEqual(sem.acquire(), True)

self.assertReturnsIfImplemented(1, get_value, sem)

self.assertEqual(sem.acquire(), True)

self.assertReturnsIfImplemented(0, get_value, sem)

self.assertEqual(sem.acquire(False), False)

self.assertReturnsIfImplemented(0, get_value, sem)

self.assertEqual(sem.release(), None)

self.assertReturnsIfImplemented(1, get_value, sem)

self.assertEqual(sem.release(), None)

self.assertReturnsIfImplemented(2, get_value, sem)

def test_semaphore(self):

sem = self.Semaphore(2)

self._test_semaphore(sem)

self.assertEqual(sem.release(), None)

self.assertReturnsIfImplemented(3, get_value, sem)

self.assertEqual(sem.release(), None)

self.assertReturnsIfImplemented(4, get_value, sem)

def test_bounded_semaphore(self):

sem = self.BoundedSemaphore(2)

self._test_semaphore(sem)

# Currently fails on OS/X

#if HAVE_GETVALUE:

# self.assertRaises(ValueError, sem.release)

# self.assertReturnsIfImplemented(2, get_value, sem)

def test_timeout(self):

if self.TYPE != 'processes':

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

sem = self.Semaphore(0)

acquire = TimingWrapper(sem.acquire)

self.assertEqual(acquire(False), False)

self.assertTimingAlmostEqual(acquire.elapsed, 0.0)

self.assertEqual(acquire(False, None), False)

self.assertTimingAlmostEqual(acquire.elapsed, 0.0)

self.assertEqual(acquire(False, TIMEOUT1), False)

self.assertTimingAlmostEqual(acquire.elapsed, 0)

self.assertEqual(acquire(True, TIMEOUT2), False)

self.assertTimingAlmostEqual(acquire.elapsed, TIMEOUT2)

self.assertEqual(acquire(timeout=TIMEOUT3), False)

@classmethod

def f(cls, cond, sleeping, woken, timeout=None):

cond.acquire()

sleeping.release()

cond.wait(timeout)

woken.release()

cond.release()

def check_invariant(self, cond):

# this is only supposed to succeed when there are no sleepers

if self.TYPE == 'processes':

try:

sleepers = (cond._sleeping_count.get_value() -

cond._woken_count.get_value())

self.assertEqual(sleepers, 0)

self.assertEqual(cond._wait_semaphore.get_value(), 0)

except NotImplementedError:

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

p = threading.Thread(target=self.f, args=(cond, sleeping, woken))

p.daemon = True

p.start()

# wait for both children to start sleeping

sleeping.acquire()

sleeping.acquire()

# check no process/thread has woken up

time.sleep(DELTA)

self.assertReturnsIfImplemented(0, get_value, woken)

# wake up one process/thread

cond.acquire()

cond.notify()

cond.release()

# check one process/thread has woken up

time.sleep(DELTA)

self.assertReturnsIfImplemented(1, get_value, woken)

# wake up another

cond.acquire()

cond.notify()

cond.release()

# check other has woken up

time.sleep(DELTA)

self.assertReturnsIfImplemented(2, get_value, woken)

# check state is not mucked up

self.check_invariant(cond)

p.join()

def test_notify_all(self):

cond = self.Condition()

sleeping = self.Semaphore(0)

woken = self.Semaphore(0)

# start some threads/processes which will timeout

for i in range(3):

p = self.Process(target=self.f,

args=(cond, sleeping, woken, TIMEOUT1))

p.daemon = True

p.start()

t = threading.Thread(target=self.f,

args=(cond, sleeping, woken, TIMEOUT1))

t.daemon = True

t.start()

# wait for them all to sleep

for i in range(6):

sleeping.acquire()

# check they have all timed out

for i in range(6):

woken.acquire()

self.assertReturnsIfImplemented(0, get_value, woken)

# check state is not mucked up

self.check_invariant(cond)

# start some more threads/processes

for i in range(3):

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 them to all sleep

for i in range(6):

sleeping.acquire()

# check no process/thread has woken up

time.sleep(DELTA)

self.assertReturnsIfImplemented(0, get_value, woken)

# wake them all up

cond.acquire()

cond.notify_all()

cond.release()

# check they have all woken

for i in range(10):

try:

if get_value(woken) == 6:

break

except NotImplementedError:

break

time.sleep(DELTA)

self.assertReturnsIfImplemented(6, get_value, woken)

# check state is not mucked up

self.check_invariant(cond)

def test_timeout(self):

cond = self.Condition()

wait = TimingWrapper(cond.wait)

cond.acquire()

res = wait(TIMEOUT1)

cond.release()

self.assertEqual(res, False)

self.assertTimingAlmostEqual(wait.elapsed, TIMEOUT1)

@classmethod

def _test_waitfor_f(cls, cond, state):

with cond:

state.value = 0

cond.notify()

result = cond.wait_for(lambda : state.value==4)

if not result or state.value != 4:

sys.exit(1)

@unittest.skipUnless(HAS_SHAREDCTYPES, 'needs sharedctypes')

def test_waitfor(self):

# based on test in test/lock_tests.py

cond = self.Condition()

state = self.Value('i', -1)

p = self.Process(target=self._test_waitfor_f, args=(cond, state))

p.daemon = True

p.start()

with cond:

result = cond.wait_for(lambda : state.value==0)

self.assertTrue(result)

self.assertEqual(state.value, 0)

for i in range(4):

time.sleep(0.01)