def__init__(self,kind,value,maxvalue,*,ctx):

ifctxisNone:

ctx=context._default_context.get_context()

name=ctx.get_start_method()

unlink_now=sys.platform=='win32'orname=='fork'

self.is_fork_ctx=ctx.get_start_method()=='fork'

unlink_now=sys.platform=='win32'orself.is_fork_ctx

foriinrange(100):

try:

sl=self._semlock=_multiprocessing.SemLock(

ifsys.platform=='win32':

h=context.get_spawning_popen().duplicate_for_child(sl.handle)

else:

ifself.is_fork_ctx:

raiseRuntimeError('A SemLock created in a fork context is being '

'multiprocessing objects and Process.')

h=sl.handle

return (h,sl.kind,sl.maxvalue,sl.name)

print(err)

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

deftest_mixed_startmethod(self):

forprocess_methodin ["spawn","forkserver"]:

queue=multiprocessing.get_context("fork").Queue()

process_ctx=multiprocessing.get_context(process_method)

p=process_ctx.Process(target=close_queue,args=(queue,))

withself.assertRaisesRegex(RuntimeError,err_msg):

p.start()

forqueue_methodin ["spawn","forkserver"]:

forprocess_methodinmultiprocessing.get_all_start_methods():

queue=multiprocessing.get_context(queue_method).Queue()

process_ctx=multiprocessing.get_context(process_method)

p=process_ctx.Process(target=close_queue,args=(queue,))

p.start()

p.join()

Ensure that multiprocessing synchronization objects created in a fork context are not sent to a different process created in a spawn context. This changes a segfault into an actionable RuntimeError in the parent process.