class DummyProcess(threading.Thread):
- def __init__(self, group=None, target=None, name=None, args=(), kwargs={}):
+ def __init__(self, group=None, target=None, name=None, args=(), kwargs=None):
threading.Thread.__init__(self, group, target, name, args, kwargs)
self._pid = None
self._children = weakref.WeakKeyDictionary()
def _Popen(self):
raise NotImplementedError
- def __init__(self, group=None, target=None, name=None, args=(), kwargs={},
+ def __init__(self, group=None, target=None, name=None, args=(), kwargs=None,
*, daemon=None):
assert group is None, 'group argument must be None for now'
count = next(_process_counter)
self._closed = False
self._target = target
self._args = tuple(args)
- self._kwargs = dict(kwargs)
+ self._kwargs = dict(kwargs) if kwargs else {}
self._name = name or type(self).__name__ + '-' + \
':'.join(str(i) for i in self._identity)
if daemon is not None:
multiprocessing.connection.Connection, -1)
+#
+# Regression tests for BaseProcess kwargs handling
+#
+
+class TestBaseProcessKwargs(unittest.TestCase):
+ def test_default_kwargs_not_shared_between_instances(self):
+ # Creating multiple Process instances without passing kwargs
+ # must create independent empty dicts (no shared state).
+ p1 = multiprocessing.Process(target=lambda: None)
+ p2 = multiprocessing.Process(target=lambda: None)
+ self.assertIsInstance(p1._kwargs, dict)
+ self.assertIsInstance(p2._kwargs, dict)
+ self.assertIsNot(p1._kwargs, p2._kwargs)
+ # Mutating one should not affect the other
+ p1._kwargs['x'] = 1
+ self.assertNotIn('x', p2._kwargs)
+
@hashlib_helper.requires_hashdigest('sha256')
class OtherTest(unittest.TestCase):
--- /dev/null
+:class:`!multiprocessing.BaseProcess` defaults ``kwargs`` to ``None`` instead of a shared dictionary.