Back to index

python3.2  3.2.2
Public Member Functions | Public Attributes | Static Public Attributes
test.test_multiprocessing._TestHeap Class Reference
Inheritance diagram for test.test_multiprocessing._TestHeap:
Inheritance graph
[legend]
Collaboration diagram for test.test_multiprocessing._TestHeap:
Collaboration graph
[legend]

List of all members.

Public Member Functions

def test_heap
def test_free_from_gc
def assertTimingAlmostEqual
def assertReturnsIfImplemented
def __reduce__

Public Attributes

_PyObject_HEAD_EXTRA Py_ssize_t ob_refcnt
struct _typeobjectob_type

Static Public Attributes

tuple ALLOWED_TYPES = ('processes',)

Detailed Description

Definition at line 1651 of file test_multiprocessing.py.


Member Function Documentation

def test.test_multiprocessing.BaseTestCase.__reduce__ (   self,
  args 
) [inherited]

Definition at line 124 of file test_multiprocessing.py.

00124 
00125     def __reduce__(self, *args):
00126         raise NotImplementedError("shouldn't try to pickle a test case")

def test.test_multiprocessing.BaseTestCase.assertReturnsIfImplemented (   self,
  value,
  func,
  args 
) [inherited]

Definition at line 114 of file test_multiprocessing.py.

00114 
00115     def assertReturnsIfImplemented(self, value, func, *args):
00116         try:
00117             res = func(*args)
00118         except NotImplementedError:
00119             pass
00120         else:
00121             return self.assertEqual(value, res)

Here is the call graph for this function:

Here is the caller graph for this function:

def test.test_multiprocessing.BaseTestCase.assertTimingAlmostEqual (   self,
  a,
  b 
) [inherited]

Definition at line 110 of file test_multiprocessing.py.

00110 
00111     def assertTimingAlmostEqual(self, a, b):
00112         if CHECK_TIMINGS:
00113             self.assertAlmostEqual(a, b, 1)

Here is the call graph for this function:

Here is the caller graph for this function:

Definition at line 1694 of file test_multiprocessing.py.

01694 
01695     def test_free_from_gc(self):
01696         # Check that freeing of blocks by the garbage collector doesn't deadlock
01697         # (issue #12352).
01698         # Make sure the GC is enabled, and set lower collection thresholds to
01699         # make collections more frequent (and increase the probability of
01700         # deadlock).
01701         if not gc.isenabled():
01702             gc.enable()
01703             self.addCleanup(gc.disable)
01704         thresholds = gc.get_threshold()
01705         self.addCleanup(gc.set_threshold, *thresholds)
01706         gc.set_threshold(10)
01707 
01708         # perform numerous block allocations, with cyclic references to make
01709         # sure objects are collected asynchronously by the gc
01710         for i in range(5000):
01711             a = multiprocessing.heap.BufferWrapper(1)
01712             b = multiprocessing.heap.BufferWrapper(1)
01713             # circular references
01714             a.buddy = b
01715             b.buddy = a
01716 
01717 #
01718 #
01719 #

Here is the call graph for this function:

Definition at line 1655 of file test_multiprocessing.py.

01655 
01656     def test_heap(self):
01657         iterations = 5000
01658         maxblocks = 50
01659         blocks = []
01660 
01661         # create and destroy lots of blocks of different sizes
01662         for i in range(iterations):
01663             size = int(random.lognormvariate(0, 1) * 1000)
01664             b = multiprocessing.heap.BufferWrapper(size)
01665             blocks.append(b)
01666             if len(blocks) > maxblocks:
01667                 i = random.randrange(maxblocks)
01668                 del blocks[i]
01669 
01670         # get the heap object
01671         heap = multiprocessing.heap.BufferWrapper._heap
01672 
01673         # verify the state of the heap
01674         all = []
01675         occupied = 0
01676         heap._lock.acquire()
01677         self.addCleanup(heap._lock.release)
01678         for L in list(heap._len_to_seq.values()):
01679             for arena, start, stop in L:
01680                 all.append((heap._arenas.index(arena), start, stop,
01681                             stop-start, 'free'))
01682         for arena, start, stop in heap._allocated_blocks:
01683             all.append((heap._arenas.index(arena), start, stop,
01684                         stop-start, 'occupied'))
01685             occupied += (stop-start)
01686 
01687         all.sort()
01688 
01689         for i in range(len(all)-1):
01690             (arena, start, stop) = all[i][:3]
01691             (narena, nstart, nstop) = all[i+1][:3]
01692             self.assertTrue((arena != narena and nstart == 0) or
01693                             (stop == nstart))

Here is the call graph for this function:


Member Data Documentation

tuple test.test_multiprocessing._TestHeap.ALLOWED_TYPES = ('processes',) [static]

Reimplemented from test.test_multiprocessing.BaseTestCase.

Definition at line 1653 of file test_multiprocessing.py.

Definition at line 107 of file object.h.

struct _typeobject* _object::ob_type [inherited]

Definition at line 108 of file object.h.


The documentation for this class was generated from the following file: