Issue #12352: Fix a deadlock in multiprocessing.Heap when a block is freed by
the garbage collector while the Heap lock is held.
This commit is contained in:
parent
ff48c0a89b
commit
414d0faedc
|
@ -101,6 +101,8 @@ class Heap(object):
|
||||||
self._stop_to_block = {}
|
self._stop_to_block = {}
|
||||||
self._allocated_blocks = set()
|
self._allocated_blocks = set()
|
||||||
self._arenas = []
|
self._arenas = []
|
||||||
|
# list of pending blocks to free - see free() comment below
|
||||||
|
self._pending_free_blocks = []
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def _roundup(n, alignment):
|
def _roundup(n, alignment):
|
||||||
|
@ -175,15 +177,39 @@ class Heap(object):
|
||||||
|
|
||||||
return start, stop
|
return start, stop
|
||||||
|
|
||||||
def free(self, block):
|
def _free_pending_blocks(self):
|
||||||
# free a block returned by malloc()
|
# Free all the blocks in the pending list - called with the lock held.
|
||||||
assert os.getpid() == self._lastpid
|
while True:
|
||||||
self._lock.acquire()
|
try:
|
||||||
try:
|
block = self._pending_free_blocks.pop()
|
||||||
|
except IndexError:
|
||||||
|
break
|
||||||
self._allocated_blocks.remove(block)
|
self._allocated_blocks.remove(block)
|
||||||
self._free(block)
|
self._free(block)
|
||||||
finally:
|
|
||||||
self._lock.release()
|
def free(self, block):
|
||||||
|
# free a block returned by malloc()
|
||||||
|
# Since free() can be called asynchronously by the GC, it could happen
|
||||||
|
# that it's called while self._lock is held: in that case,
|
||||||
|
# self._lock.acquire() would deadlock (issue #12352). To avoid that, a
|
||||||
|
# trylock is used instead, and if the lock can't be acquired
|
||||||
|
# immediately, the block is added to a list of blocks to be freed
|
||||||
|
# synchronously sometimes later from malloc() or free(), by calling
|
||||||
|
# _free_pending_blocks() (appending and retrieving from a list is not
|
||||||
|
# strictly thread-safe but under cPython it's atomic thanks to the GIL).
|
||||||
|
assert os.getpid() == self._lastpid
|
||||||
|
if not self._lock.acquire(False):
|
||||||
|
# can't acquire the lock right now, add the block to the list of
|
||||||
|
# pending blocks to free
|
||||||
|
self._pending_free_blocks.append(block)
|
||||||
|
else:
|
||||||
|
# we hold the lock
|
||||||
|
try:
|
||||||
|
self._free_pending_blocks()
|
||||||
|
self._allocated_blocks.remove(block)
|
||||||
|
self._free(block)
|
||||||
|
finally:
|
||||||
|
self._lock.release()
|
||||||
|
|
||||||
def malloc(self, size):
|
def malloc(self, size):
|
||||||
# return a block of right size (possibly rounded up)
|
# return a block of right size (possibly rounded up)
|
||||||
|
@ -191,6 +217,7 @@ class Heap(object):
|
||||||
if os.getpid() != self._lastpid:
|
if os.getpid() != self._lastpid:
|
||||||
self.__init__() # reinitialize after fork
|
self.__init__() # reinitialize after fork
|
||||||
self._lock.acquire()
|
self._lock.acquire()
|
||||||
|
self._free_pending_blocks()
|
||||||
try:
|
try:
|
||||||
size = self._roundup(max(size,1), self._alignment)
|
size = self._roundup(max(size,1), self._alignment)
|
||||||
(arena, start, stop) = self._malloc(size)
|
(arena, start, stop) = self._malloc(size)
|
||||||
|
|
|
@ -1615,6 +1615,8 @@ class _TestHeap(BaseTestCase):
|
||||||
# verify the state of the heap
|
# verify the state of the heap
|
||||||
all = []
|
all = []
|
||||||
occupied = 0
|
occupied = 0
|
||||||
|
heap._lock.acquire()
|
||||||
|
self.addCleanup(heap._lock.release)
|
||||||
for L in heap._len_to_seq.values():
|
for L in heap._len_to_seq.values():
|
||||||
for arena, start, stop in L:
|
for arena, start, stop in L:
|
||||||
all.append((heap._arenas.index(arena), start, stop,
|
all.append((heap._arenas.index(arena), start, stop,
|
||||||
|
@ -1632,6 +1634,29 @@ class _TestHeap(BaseTestCase):
|
||||||
self.assertTrue((arena != narena and nstart == 0) or
|
self.assertTrue((arena != narena and nstart == 0) or
|
||||||
(stop == nstart))
|
(stop == nstart))
|
||||||
|
|
||||||
|
def test_free_from_gc(self):
|
||||||
|
# Check that freeing of blocks by the garbage collector doesn't deadlock
|
||||||
|
# (issue #12352).
|
||||||
|
# Make sure the GC is enabled, and set lower collection thresholds to
|
||||||
|
# make collections more frequent (and increase the probability of
|
||||||
|
# deadlock).
|
||||||
|
if gc.isenabled():
|
||||||
|
thresholds = gc.get_threshold()
|
||||||
|
self.addCleanup(gc.set_threshold, *thresholds)
|
||||||
|
else:
|
||||||
|
gc.enable()
|
||||||
|
self.addCleanup(gc.disable)
|
||||||
|
gc.set_threshold(10)
|
||||||
|
|
||||||
|
# perform numerous block allocations, with cyclic references to make
|
||||||
|
# sure objects are collected asynchronously by the gc
|
||||||
|
for i in range(5000):
|
||||||
|
a = multiprocessing.heap.BufferWrapper(1)
|
||||||
|
b = multiprocessing.heap.BufferWrapper(1)
|
||||||
|
# circular references
|
||||||
|
a.buddy = b
|
||||||
|
b.buddy = a
|
||||||
|
|
||||||
#
|
#
|
||||||
#
|
#
|
||||||
#
|
#
|
||||||
|
|
|
@ -18,6 +18,9 @@ Core and Builtins
|
||||||
Library
|
Library
|
||||||
-------
|
-------
|
||||||
|
|
||||||
|
- Issue #12352: Fix a deadlock in multiprocessing.Heap when a block is freed by
|
||||||
|
the garbage collector while the Heap lock is held.
|
||||||
|
|
||||||
- Issue #9516: On Mac OS X, change Distutils to no longer globally attempt to
|
- Issue #9516: On Mac OS X, change Distutils to no longer globally attempt to
|
||||||
check or set the MACOSX_DEPLOYMENT_TARGET environment variable for the
|
check or set the MACOSX_DEPLOYMENT_TARGET environment variable for the
|
||||||
interpreter process. This could cause failures in non-Distutils subprocesses
|
interpreter process. This could cause failures in non-Distutils subprocesses
|
||||||
|
|
Loading…
Reference in New Issue