cpython: fd8dc3746992 (original) (raw)

Mercurial > cpython

changeset 71128:fd8dc3746992

Merge issue #12352: Fix a deadlock in multiprocessing.Heap when a block is freed by the garbage collector while the Heap lock is held. [#12352]

Charles-François Natali neologix@free.fr
date	Sat, 02 Jul 2011 14:43:11 +0200
parents	775356b583d1(current diff)37606505b227(diff)
children	6b3872a11299
files	Lib/test/test_multiprocessing.py Misc/NEWS
diffstat	3 files changed, 60 insertions(+), 6 deletions(-)[+] [-] Lib/multiprocessing/heap.py 39 Lib/test/test_multiprocessing.py 24 Misc/NEWS 3

line wrap: on

line diff

--- a/Lib/multiprocessing/heap.py +++ b/Lib/multiprocessing/heap.py @@ -101,6 +101,8 @@ class Heap(object): self._stop_to_block = {} self._allocated_blocks = set() self._arenas = []

   # list of pending blocks to free - see free() comment below[](#l1.7)

   self._pending_free_blocks = [][](#l1.8)

@staticmethod def _roundup(n, alignment): @@ -175,15 +177,39 @@ class Heap(object): return start, stop

def _free_pending_blocks(self):

   # Free all the blocks in the pending list - called with the lock held.[](#l1.17)

```
   while True:[](#l1.18)
```
```
       try:[](#l1.19)
```

           block = self._pending_free_blocks.pop()[](#l1.20)

```
       except IndexError:[](#l1.21)
```
```
           break[](#l1.22)
```

       self._allocated_blocks.remove(block)[](#l1.23)

```
       self._free(block)[](#l1.24)
```

+ def free(self, block): # free a block returned by malloc()

   # Since free() can be called asynchronously by the GC, it could happen[](#l1.28)

   # that it's called while self._lock is held: in that case,[](#l1.29)

   # self._lock.acquire() would deadlock (issue #12352). To avoid that, a[](#l1.30)

   # trylock is used instead, and if the lock can't be acquired[](#l1.31)

   # immediately, the block is added to a list of blocks to be freed[](#l1.32)

   # synchronously sometimes later from malloc() or free(), by calling[](#l1.33)

   # _free_pending_blocks() (appending and retrieving from a list is not[](#l1.34)

   # strictly thread-safe but under cPython it's atomic thanks to the GIL).[](#l1.35)
   assert os.getpid() == self._lastpid[](#l1.36)

```
   self._lock.acquire()[](#l1.37)
```
```
   try:[](#l1.38)
```

       self._allocated_blocks.remove(block)[](#l1.39)

```
       self._free(block)[](#l1.40)
```
```
   finally:[](#l1.41)
```
```
       self._lock.release()[](#l1.42)
```

   if not self._lock.acquire(False):[](#l1.43)

       # can't acquire the lock right now, add the block to the list of[](#l1.44)

       # pending blocks to free[](#l1.45)

       self._pending_free_blocks.append(block)[](#l1.46)

```
   else:[](#l1.47)
```
```
       # we hold the lock[](#l1.48)
```
```
       try:[](#l1.49)
```

           self._free_pending_blocks()[](#l1.50)

           self._allocated_blocks.remove(block)[](#l1.51)

```
           self._free(block)[](#l1.52)
```
```
       finally:[](#l1.53)
```

           self._lock.release()[](#l1.54)

def malloc(self, size): # return a block of right size (possibly rounded up) @@ -191,6 +217,7 @@ class Heap(object): if os.getpid() != self._lastpid: self.init() # reinitialize after fork self._lock.acquire()

   self._free_pending_blocks()[](#l1.62)
   try:[](#l1.63)
       size = self._roundup(max(size,1), self._alignment)[](#l1.64)
       (arena, start, stop) = self._malloc(size)[](#l1.65)

--- a/Lib/test/test_multiprocessing.py +++ b/Lib/test/test_multiprocessing.py @@ -1721,6 +1721,8 @@ class _TestHeap(BaseTestCase): # verify the state of the heap all = [] occupied = 0

```
   heap._lock.acquire()[](#l2.7)
```

   self.addCleanup(heap._lock.release)[](#l2.8)
   for L in list(heap._len_to_seq.values()):[](#l2.9)
       for arena, start, stop in L:[](#l2.10)
           all.append((heap._arenas.index(arena), start, stop,[](#l2.11)

@@ -1738,6 +1740,28 @@ class _TestHeap(BaseTestCase): self.assertTrue((arena != narena and nstart == 0) or (stop == nstart))

def test_free_from_gc(self):

   # Check that freeing of blocks by the garbage collector doesn't deadlock[](#l2.17)

```
   # (issue #12352).[](#l2.18)
```

   # Make sure the GC is enabled, and set lower collection thresholds to[](#l2.19)

   # make collections more frequent (and increase the probability of[](#l2.20)

```
   # deadlock).[](#l2.21)
```
```
   if not gc.isenabled():[](#l2.22)
```
```
       gc.enable()[](#l2.23)
```

       self.addCleanup(gc.disable)[](#l2.24)

   thresholds = gc.get_threshold()[](#l2.25)

   self.addCleanup(gc.set_threshold, *thresholds)[](#l2.26)

```
   gc.set_threshold(10)[](#l2.27)
```

   # perform numerous block allocations, with cyclic references to make[](#l2.29)

   # sure objects are collected asynchronously by the gc[](#l2.30)

```
   for i in range(5000):[](#l2.31)
```

       a = multiprocessing.heap.BufferWrapper(1)[](#l2.32)

       b = multiprocessing.heap.BufferWrapper(1)[](#l2.33)

```
       # circular references[](#l2.34)
```
```
       a.buddy = b[](#l2.35)
```
```
       b.buddy = a[](#l2.36)
```

+ # # #

--- a/Misc/NEWS +++ b/Misc/NEWS @@ -200,6 +200,9 @@ Core and Builtins 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 #12462: time.sleep() now calls immediatly the (Python) signal handler if it is interrupted by a signal, instead of having to wait until the next instruction.