heap.py 9.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289
  1. #
  2. # Module which supports allocation of memory from an mmap
  3. #
  4. # multiprocessing/heap.py
  5. #
  6. # Copyright (c) 2006-2008, R Oudkerk
  7. # Licensed to PSF under a Contributor Agreement.
  8. #
  9. from __future__ import absolute_import
  10. import bisect
  11. import errno
  12. import io
  13. import mmap
  14. import os
  15. import sys
  16. import threading
  17. import tempfile
  18. from . import context
  19. from . import reduction
  20. from . import util
  21. from ._ext import _billiard, win32
  22. __all__ = ['BufferWrapper']
  23. PY3 = sys.version_info[0] == 3
  24. #
  25. # Inheritable class which wraps an mmap, and from which blocks can be allocated
  26. #
  27. if sys.platform == 'win32':
  28. class Arena(object):
  29. _rand = tempfile._RandomNameSequence()
  30. def __init__(self, size):
  31. self.size = size
  32. for i in range(100):
  33. name = 'pym-%d-%s' % (os.getpid(), next(self._rand))
  34. buf = mmap.mmap(-1, size, tagname=name)
  35. if win32.GetLastError() == 0:
  36. break
  37. # we have reopened a preexisting map
  38. buf.close()
  39. else:
  40. exc = IOError('Cannot find name for new mmap')
  41. exc.errno = errno.EEXIST
  42. raise exc
  43. self.name = name
  44. self.buffer = buf
  45. self._state = (self.size, self.name)
  46. def __getstate__(self):
  47. context.assert_spawning(self)
  48. return self._state
  49. def __setstate__(self, state):
  50. self.size, self.name = self._state = state
  51. self.buffer = mmap.mmap(-1, self.size, tagname=self.name)
  52. # XXX Temporarily preventing buildbot failures while determining
  53. # XXX the correct long-term fix. See issue #23060
  54. # assert win32.GetLastError() == win32.ERROR_ALREADY_EXISTS
  55. else:
  56. class Arena(object):
  57. def __init__(self, size, fd=-1):
  58. self.size = size
  59. self.fd = fd
  60. if fd == -1:
  61. if PY3:
  62. self.fd, name = tempfile.mkstemp(
  63. prefix='pym-%d-' % (os.getpid(),),
  64. dir=util.get_temp_dir(),
  65. )
  66. os.unlink(name)
  67. util.Finalize(self, os.close, (self.fd,))
  68. with io.open(self.fd, 'wb', closefd=False) as f:
  69. bs = 1024 * 1024
  70. if size >= bs:
  71. zeros = b'\0' * bs
  72. for _ in range(size // bs):
  73. f.write(zeros)
  74. del(zeros)
  75. f.write(b'\0' * (size % bs))
  76. assert f.tell() == size
  77. else:
  78. name = tempfile.mktemp(
  79. prefix='pym-%d-' % (os.getpid(),),
  80. dir=util.get_temp_dir(),
  81. )
  82. self.fd = os.open(
  83. name, os.O_RDWR | os.O_CREAT | os.O_EXCL, 0o600,
  84. )
  85. util.Finalize(self, os.close, (self.fd,))
  86. os.unlink(name)
  87. os.ftruncate(self.fd, size)
  88. self.buffer = mmap.mmap(self.fd, self.size)
  89. def reduce_arena(a):
  90. if a.fd == -1:
  91. raise ValueError('Arena is unpicklable because'
  92. 'forking was enabled when it was created')
  93. return rebuild_arena, (a.size, reduction.DupFd(a.fd))
  94. def rebuild_arena(size, dupfd):
  95. return Arena(size, dupfd.detach())
  96. reduction.register(Arena, reduce_arena)
  97. #
  98. # Class allowing allocation of chunks of memory from arenas
  99. #
  100. class Heap(object):
  101. _alignment = 8
  102. def __init__(self, size=mmap.PAGESIZE):
  103. self._lastpid = os.getpid()
  104. self._lock = threading.Lock()
  105. self._size = size
  106. self._lengths = []
  107. self._len_to_seq = {}
  108. self._start_to_block = {}
  109. self._stop_to_block = {}
  110. self._allocated_blocks = set()
  111. self._arenas = []
  112. # list of pending blocks to free - see free() comment below
  113. self._pending_free_blocks = []
  114. @staticmethod
  115. def _roundup(n, alignment):
  116. # alignment must be a power of 2
  117. mask = alignment - 1
  118. return (n + mask) & ~mask
  119. def _malloc(self, size):
  120. # returns a large enough block -- it might be much larger
  121. i = bisect.bisect_left(self._lengths, size)
  122. if i == len(self._lengths):
  123. length = self._roundup(max(self._size, size), mmap.PAGESIZE)
  124. self._size *= 2
  125. util.info('allocating a new mmap of length %d', length)
  126. arena = Arena(length)
  127. self._arenas.append(arena)
  128. return (arena, 0, length)
  129. else:
  130. length = self._lengths[i]
  131. seq = self._len_to_seq[length]
  132. block = seq.pop()
  133. if not seq:
  134. del self._len_to_seq[length], self._lengths[i]
  135. (arena, start, stop) = block
  136. del self._start_to_block[(arena, start)]
  137. del self._stop_to_block[(arena, stop)]
  138. return block
  139. def _free(self, block):
  140. # free location and try to merge with neighbours
  141. (arena, start, stop) = block
  142. try:
  143. prev_block = self._stop_to_block[(arena, start)]
  144. except KeyError:
  145. pass
  146. else:
  147. start, _ = self._absorb(prev_block)
  148. try:
  149. next_block = self._start_to_block[(arena, stop)]
  150. except KeyError:
  151. pass
  152. else:
  153. _, stop = self._absorb(next_block)
  154. block = (arena, start, stop)
  155. length = stop - start
  156. try:
  157. self._len_to_seq[length].append(block)
  158. except KeyError:
  159. self._len_to_seq[length] = [block]
  160. bisect.insort(self._lengths, length)
  161. self._start_to_block[(arena, start)] = block
  162. self._stop_to_block[(arena, stop)] = block
  163. def _absorb(self, block):
  164. # deregister this block so it can be merged with a neighbour
  165. (arena, start, stop) = block
  166. del self._start_to_block[(arena, start)]
  167. del self._stop_to_block[(arena, stop)]
  168. length = stop - start
  169. seq = self._len_to_seq[length]
  170. seq.remove(block)
  171. if not seq:
  172. del self._len_to_seq[length]
  173. self._lengths.remove(length)
  174. return start, stop
  175. def _free_pending_blocks(self):
  176. # Free all the blocks in the pending list - called with the lock held
  177. while 1:
  178. try:
  179. block = self._pending_free_blocks.pop()
  180. except IndexError:
  181. break
  182. self._allocated_blocks.remove(block)
  183. self._free(block)
  184. def free(self, block):
  185. # free a block returned by malloc()
  186. # Since free() can be called asynchronously by the GC, it could happen
  187. # that it's called while self._lock is held: in that case,
  188. # self._lock.acquire() would deadlock (issue #12352). To avoid that, a
  189. # trylock is used instead, and if the lock can't be acquired
  190. # immediately, the block is added to a list of blocks to be freed
  191. # synchronously sometimes later from malloc() or free(), by calling
  192. # _free_pending_blocks() (appending and retrieving from a list is not
  193. # strictly thread-safe but under cPython it's atomic
  194. # thanks to the GIL).
  195. assert os.getpid() == self._lastpid
  196. if not self._lock.acquire(False):
  197. # can't acquire the lock right now, add the block to the list of
  198. # pending blocks to free
  199. self._pending_free_blocks.append(block)
  200. else:
  201. # we hold the lock
  202. try:
  203. self._free_pending_blocks()
  204. self._allocated_blocks.remove(block)
  205. self._free(block)
  206. finally:
  207. self._lock.release()
  208. def malloc(self, size):
  209. # return a block of right size (possibly rounded up)
  210. assert 0 <= size < sys.maxsize
  211. if os.getpid() != self._lastpid:
  212. self.__init__() # reinitialize after fork
  213. with self._lock:
  214. self._free_pending_blocks()
  215. size = self._roundup(max(size, 1), self._alignment)
  216. (arena, start, stop) = self._malloc(size)
  217. new_stop = start + size
  218. if new_stop < stop:
  219. self._free((arena, new_stop, stop))
  220. block = (arena, start, new_stop)
  221. self._allocated_blocks.add(block)
  222. return block
  223. #
  224. # Class representing a chunk of an mmap -- can be inherited
  225. #
  226. class BufferWrapper(object):
  227. _heap = Heap()
  228. def __init__(self, size):
  229. assert 0 <= size < sys.maxsize
  230. block = BufferWrapper._heap.malloc(size)
  231. self._state = (block, size)
  232. util.Finalize(self, BufferWrapper._heap.free, args=(block,))
  233. def get_address(self):
  234. (arena, start, stop), size = self._state
  235. address, length = _billiard.address_of_buffer(arena.buffer)
  236. assert size <= length
  237. return address + start
  238. def get_size(self):
  239. return self._state[1]
  240. def create_memoryview(self):
  241. (arena, start, stop), size = self._state
  242. return memoryview(arena.buffer)[start:start + size]