A shared memory area backed by anonymous memory (Windows).
| 29 | import _winapi |
| 30 | |
| 31 | class Arena(object): |
| 32 | """ |
| 33 | A shared memory area backed by anonymous memory (Windows). |
| 34 | """ |
| 35 | |
| 36 | _rand = tempfile._RandomNameSequence() |
| 37 | |
| 38 | def __init__(self, size): |
| 39 | self.size = size |
| 40 | for i in range(100): |
| 41 | name = 'pym-%d-%s' % (os.getpid(), next(self._rand)) |
| 42 | buf = mmap.mmap(-1, size, tagname=name) |
| 43 | if _winapi.GetLastError() == 0: |
| 44 | break |
| 45 | # We have reopened a preexisting mmap. |
| 46 | buf.close() |
| 47 | else: |
| 48 | raise FileExistsError('Cannot find name for new mmap') |
| 49 | self.name = name |
| 50 | self.buffer = buf |
| 51 | self._state = (self.size, self.name) |
| 52 | |
| 53 | def __getstate__(self): |
| 54 | assert_spawning(self) |
| 55 | return self._state |
| 56 | |
| 57 | def __setstate__(self, state): |
| 58 | self.size, self.name = self._state = state |
| 59 | # Reopen existing mmap |
| 60 | self.buffer = mmap.mmap(-1, self.size, tagname=self.name) |
| 61 | # XXX Temporarily preventing buildbot failures while determining |
| 62 | # XXX the correct long-term fix. See issue 23060 |
| 63 | #assert _winapi.GetLastError() == _winapi.ERROR_ALREADY_EXISTS |
| 64 | |
| 65 | else: |
| 66 |
no outgoing calls
no test coverage detected
searching dependent graphs…