import unittest

import queue as pyqueue

import contextlib

import time

import io

import itertools

self.fail("failed spawning forkserver or grandchild")

#

# Check that killing process does not leak named semaphores

#

@unittest.skipIf(sys.platform == "win32",

"test semantics don't make sense on Windows")

class TestSemaphoreTracker(unittest.TestCase):

def test_semaphore_tracker(self):

#

# Check that killing process does not leak named semaphores

#

import subprocess

cmd = '''if 1:

import multiprocessing as mp, time, os

self.assertRegex(err, expected)

self.assertRegex(err, r'semaphore_tracker: %r: \[Errno' % name1)

def check_semaphore_tracker_death(self, signum, should_die):

# bpo-31310: if the semaphore tracker process has died, it should

# be restarted implicitly.

from multiprocessing.semaphore_tracker import _semaphore_tracker

_semaphore_tracker.ensure_running()

pid = _semaphore_tracker._pid

os.kill(pid, signum)

time.sleep(1.0) # give it time to die

ctx = multiprocessing.get_context("spawn")

with contextlib.ExitStack() as stack:

if should_die:

stack.enter_context(self.assertWarnsRegex(

UserWarning,

"semaphore_tracker: process died"))

sem = ctx.Semaphore()

sem.acquire()

sem.release()

wr = weakref.ref(sem)

# ensure `sem` gets collected, which triggers communication with

# the semaphore tracker

del sem

gc.collect()

self.assertIsNone(wr())

def test_semaphore_tracker_sigint(self):

# Catchable signal (ignored by semaphore tracker)

self.check_semaphore_tracker_death(signal.SIGINT, False)

def test_semaphore_tracker_sigkill(self):

# Uncatchable signal.

self.check_semaphore_tracker_death(signal.SIGKILL, True)

class TestSimpleQueue(unittest.TestCase):

@classmethod