Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/joblib/_memmapping_reducer.py: 23%

Shortcuts on this page

r m x   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

254 statements  

1""" 

2Reducer using memory mapping for numpy arrays 

3""" 

4# Author: Thomas Moreau <thomas.moreau.2010@gmail.com> 

5# Copyright: 2017, Thomas Moreau 

6# License: BSD 3 clause 

7 

8from mmap import mmap 

9import errno 

10import os 

11import stat 

12import threading 

13import atexit 

14import tempfile 

15import time 

16import warnings 

17import weakref 

18from uuid import uuid4 

19from multiprocessing import util 

20 

21from pickle import whichmodule, loads, dumps, HIGHEST_PROTOCOL, PicklingError 

22 

23try: 

24 WindowsError 

25except NameError: 

26 WindowsError = type(None) 

27 

28try: 

29 import numpy as np 

30 from numpy.lib.stride_tricks import as_strided 

31except ImportError: 

32 np = None 

33 

34from .numpy_pickle import dump, load, load_temporary_memmap 

35from .backports import make_memmap 

36from .disk import delete_folder 

37from .externals.loky.backend import resource_tracker 

38 

39# Some system have a ramdisk mounted by default, we can use it instead of /tmp 

40# as the default folder to dump big arrays to share with subprocesses. 

41SYSTEM_SHARED_MEM_FS = '/dev/shm' 

42 

43# Minimal number of bytes available on SYSTEM_SHARED_MEM_FS to consider using 

44# it as the default folder to dump big arrays to share with subprocesses. 

45SYSTEM_SHARED_MEM_FS_MIN_SIZE = int(2e9) 

46 

47# Folder and file permissions to chmod temporary files generated by the 

48# memmapping pool. Only the owner of the Python process can access the 

49# temporary files and folder. 

50FOLDER_PERMISSIONS = stat.S_IRUSR | stat.S_IWUSR | stat.S_IXUSR 

51FILE_PERMISSIONS = stat.S_IRUSR | stat.S_IWUSR 

52 

53# Set used in joblib workers, referencing the filenames of temporary memmaps 

54# created by joblib to speed up data communication. In child processes, we add 

55# a finalizer to these memmaps that sends a maybe_unlink call to the 

56# resource_tracker, in order to free main memory as fast as possible. 

57JOBLIB_MMAPS = set() 

58 

59 

60def _log_and_unlink(filename): 

61 from .externals.loky.backend.resource_tracker import _resource_tracker 

62 util.debug( 

63 "[FINALIZER CALL] object mapping to {} about to be deleted," 

64 " decrementing the refcount of the file (pid: {})".format( 

65 os.path.basename(filename), os.getpid())) 

66 _resource_tracker.maybe_unlink(filename, "file") 

67 

68 

69def add_maybe_unlink_finalizer(memmap): 

70 util.debug( 

71 "[FINALIZER ADD] adding finalizer to {} (id {}, filename {}, pid {})" 

72 "".format(type(memmap), id(memmap), os.path.basename(memmap.filename), 

73 os.getpid())) 

74 weakref.finalize(memmap, _log_and_unlink, memmap.filename) 

75 

76 

77def unlink_file(filename): 

78 """Wrapper around os.unlink with a retry mechanism. 

79 

80 The retry mechanism has been implemented primarily to overcome a race 

81 condition happening during the finalizer of a np.memmap: when a process 

82 holding the last reference to a mmap-backed np.memmap/np.array is about to 

83 delete this array (and close the reference), it sends a maybe_unlink 

84 request to the resource_tracker. This request can be processed faster than 

85 it takes for the last reference of the memmap to be closed, yielding (on 

86 Windows) a PermissionError in the resource_tracker loop. 

87 """ 

88 NUM_RETRIES = 10 

89 for retry_no in range(1, NUM_RETRIES + 1): 

90 try: 

91 os.unlink(filename) 

92 break 

93 except PermissionError: 

94 util.debug( 

95 '[ResourceTracker] tried to unlink {}, got ' 

96 'PermissionError'.format(filename) 

97 ) 

98 if retry_no == NUM_RETRIES: 

99 raise 

100 else: 

101 time.sleep(.2) 

102 except FileNotFoundError: 

103 # In case of a race condition when deleting the temporary folder, 

104 # avoid noisy FileNotFoundError exception in the resource tracker. 

105 pass 

106 

107 

108resource_tracker._CLEANUP_FUNCS['file'] = unlink_file 

109 

110 

111class _WeakArrayKeyMap: 

112 """A variant of weakref.WeakKeyDictionary for unhashable numpy arrays. 

113 

114 This datastructure will be used with numpy arrays as obj keys, therefore we 

115 do not use the __get__ / __set__ methods to avoid any conflict with the 

116 numpy fancy indexing syntax. 

117 """ 

118 

119 def __init__(self): 

120 self._data = {} 

121 

122 def get(self, obj): 

123 ref, val = self._data[id(obj)] 

124 if ref() is not obj: 

125 # In case of race condition with on_destroy: could never be 

126 # triggered by the joblib tests with CPython. 

127 raise KeyError(obj) 

128 return val 

129 

130 def set(self, obj, value): 

131 key = id(obj) 

132 try: 

133 ref, _ = self._data[key] 

134 if ref() is not obj: 

135 # In case of race condition with on_destroy: could never be 

136 # triggered by the joblib tests with CPython. 

137 raise KeyError(obj) 

138 except KeyError: 

139 # Insert the new entry in the mapping along with a weakref 

140 # callback to automatically delete the entry from the mapping 

141 # as soon as the object used as key is garbage collected. 

142 def on_destroy(_): 

143 del self._data[key] 

144 ref = weakref.ref(obj, on_destroy) 

145 self._data[key] = ref, value 

146 

147 def __getstate__(self): 

148 raise PicklingError("_WeakArrayKeyMap is not pickleable") 

149 

150 

151############################################################################### 

152# Support for efficient transient pickling of numpy data structures 

153 

154 

155def _get_backing_memmap(a): 

156 """Recursively look up the original np.memmap instance base if any.""" 

157 b = getattr(a, 'base', None) 

158 if b is None: 

159 # TODO: check scipy sparse datastructure if scipy is installed 

160 # a nor its descendants do not have a memmap base 

161 return None 

162 

163 elif isinstance(b, mmap): 

164 # a is already a real memmap instance. 

165 return a 

166 

167 else: 

168 # Recursive exploration of the base ancestry 

169 return _get_backing_memmap(b) 

170 

171 

172def _get_temp_dir(pool_folder_name, temp_folder=None): 

173 """Get the full path to a subfolder inside the temporary folder. 

174 

175 Parameters 

176 ---------- 

177 pool_folder_name : str 

178 Sub-folder name used for the serialization of a pool instance. 

179 

180 temp_folder: str, optional 

181 Folder to be used by the pool for memmapping large arrays 

182 for sharing memory with worker processes. If None, this will try in 

183 order: 

184 

185 - a folder pointed by the JOBLIB_TEMP_FOLDER environment 

186 variable, 

187 - /dev/shm if the folder exists and is writable: this is a 

188 RAMdisk filesystem available by default on modern Linux 

189 distributions, 

190 - the default system temporary folder that can be 

191 overridden with TMP, TMPDIR or TEMP environment 

192 variables, typically /tmp under Unix operating systems. 

193 

194 Returns 

195 ------- 

196 pool_folder : str 

197 full path to the temporary folder 

198 use_shared_mem : bool 

199 whether the temporary folder is written to the system shared memory 

200 folder or some other temporary folder. 

201 """ 

202 use_shared_mem = False 

203 if temp_folder is None: 

204 temp_folder = os.environ.get('JOBLIB_TEMP_FOLDER', None) 

205 if temp_folder is None: 

206 if os.path.exists(SYSTEM_SHARED_MEM_FS) and hasattr(os, 'statvfs'): 

207 try: 

208 shm_stats = os.statvfs(SYSTEM_SHARED_MEM_FS) 

209 available_nbytes = shm_stats.f_bsize * shm_stats.f_bavail 

210 if available_nbytes > SYSTEM_SHARED_MEM_FS_MIN_SIZE: 

211 # Try to see if we have write access to the shared mem 

212 # folder only if it is reasonably large (that is 2GB or 

213 # more). 

214 temp_folder = SYSTEM_SHARED_MEM_FS 

215 pool_folder = os.path.join(temp_folder, pool_folder_name) 

216 if not os.path.exists(pool_folder): 

217 os.makedirs(pool_folder) 

218 use_shared_mem = True 

219 except (IOError, OSError): 

220 # Missing rights in the /dev/shm partition, fallback to regular 

221 # temp folder. 

222 temp_folder = None 

223 if temp_folder is None: 

224 # Fallback to the default tmp folder, typically /tmp 

225 temp_folder = tempfile.gettempdir() 

226 temp_folder = os.path.abspath(os.path.expanduser(temp_folder)) 

227 pool_folder = os.path.join(temp_folder, pool_folder_name) 

228 return pool_folder, use_shared_mem 

229 

230 

231def has_shareable_memory(a): 

232 """Return True if a is backed by some mmap buffer directly or not.""" 

233 return _get_backing_memmap(a) is not None 

234 

235 

236def _strided_from_memmap(filename, dtype, mode, offset, order, shape, strides, 

237 total_buffer_len, unlink_on_gc_collect): 

238 """Reconstruct an array view on a memory mapped file.""" 

239 if mode == 'w+': 

240 # Do not zero the original data when unpickling 

241 mode = 'r+' 

242 

243 if strides is None: 

244 # Simple, contiguous memmap 

245 return make_memmap( 

246 filename, dtype=dtype, shape=shape, mode=mode, offset=offset, 

247 order=order, unlink_on_gc_collect=unlink_on_gc_collect 

248 ) 

249 else: 

250 # For non-contiguous data, memmap the total enclosing buffer and then 

251 # extract the non-contiguous view with the stride-tricks API 

252 base = make_memmap( 

253 filename, dtype=dtype, shape=total_buffer_len, offset=offset, 

254 mode=mode, order=order, unlink_on_gc_collect=unlink_on_gc_collect 

255 ) 

256 return as_strided(base, shape=shape, strides=strides) 

257 

258 

259def _reduce_memmap_backed(a, m): 

260 """Pickling reduction for memmap backed arrays. 

261 

262 a is expected to be an instance of np.ndarray (or np.memmap) 

263 m is expected to be an instance of np.memmap on the top of the ``base`` 

264 attribute ancestry of a. ``m.base`` should be the real python mmap object. 

265 """ 

266 # offset that comes from the striding differences between a and m 

267 util.debug('[MEMMAP REDUCE] reducing a memmap-backed array ' 

268 '(shape, {}, pid: {})'.format(a.shape, os.getpid())) 

269 try: 

270 from numpy.lib.array_utils import byte_bounds 

271 except (ModuleNotFoundError, ImportError): 

272 # Backward-compat for numpy < 2.0 

273 from numpy import byte_bounds 

274 a_start, a_end = byte_bounds(a) 

275 m_start = byte_bounds(m)[0] 

276 offset = a_start - m_start 

277 

278 # offset from the backing memmap 

279 offset += m.offset 

280 

281 if m.flags['F_CONTIGUOUS']: 

282 order = 'F' 

283 else: 

284 # The backing memmap buffer is necessarily contiguous hence C if not 

285 # Fortran 

286 order = 'C' 

287 

288 if a.flags['F_CONTIGUOUS'] or a.flags['C_CONTIGUOUS']: 

289 # If the array is a contiguous view, no need to pass the strides 

290 strides = None 

291 total_buffer_len = None 

292 else: 

293 # Compute the total number of items to map from which the strided 

294 # view will be extracted. 

295 strides = a.strides 

296 total_buffer_len = (a_end - a_start) // a.itemsize 

297 

298 return (_strided_from_memmap, 

299 (m.filename, a.dtype, m.mode, offset, order, a.shape, strides, 

300 total_buffer_len, False)) 

301 

302 

303def reduce_array_memmap_backward(a): 

304 """reduce a np.array or a np.memmap from a child process""" 

305 m = _get_backing_memmap(a) 

306 if isinstance(m, np.memmap) and m.filename not in JOBLIB_MMAPS: 

307 # if a is backed by a memmaped file, reconstruct a using the 

308 # memmaped file. 

309 return _reduce_memmap_backed(a, m) 

310 else: 

311 # a is either a regular (not memmap-backed) numpy array, or an array 

312 # backed by a shared temporary file created by joblib. In the latter 

313 # case, in order to limit the lifespan of these temporary files, we 

314 # serialize the memmap as a regular numpy array, and decref the 

315 # file backing the memmap (done implicitly in a previously registered 

316 # finalizer, see ``unlink_on_gc_collect`` for more details) 

317 return ( 

318 loads, (dumps(np.asarray(a), protocol=HIGHEST_PROTOCOL), ) 

319 ) 

320 

321 

322class ArrayMemmapForwardReducer(object): 

323 """Reducer callable to dump large arrays to memmap files. 

324 

325 Parameters 

326 ---------- 

327 max_nbytes: int 

328 Threshold to trigger memmapping of large arrays to files created 

329 a folder. 

330 temp_folder_resolver: callable 

331 An callable in charge of resolving a temporary folder name where files 

332 for backing memmapped arrays are created. 

333 mmap_mode: 'r', 'r+' or 'c' 

334 Mode for the created memmap datastructure. See the documentation of 

335 numpy.memmap for more details. Note: 'w+' is coerced to 'r+' 

336 automatically to avoid zeroing the data on unpickling. 

337 verbose: int, optional, 0 by default 

338 If verbose > 0, memmap creations are logged. 

339 If verbose > 1, both memmap creations, reuse and array pickling are 

340 logged. 

341 prewarm: bool, optional, False by default. 

342 Force a read on newly memmapped array to make sure that OS pre-cache it 

343 memory. This can be useful to avoid concurrent disk access when the 

344 same data array is passed to different worker processes. 

345 """ 

346 

347 def __init__(self, max_nbytes, temp_folder_resolver, mmap_mode, 

348 unlink_on_gc_collect, verbose=0, prewarm=True): 

349 self._max_nbytes = max_nbytes 

350 self._temp_folder_resolver = temp_folder_resolver 

351 self._mmap_mode = mmap_mode 

352 self.verbose = int(verbose) 

353 if prewarm == "auto": 

354 self._prewarm = not self._temp_folder.startswith( 

355 SYSTEM_SHARED_MEM_FS 

356 ) 

357 else: 

358 self._prewarm = prewarm 

359 self._prewarm = prewarm 

360 self._memmaped_arrays = _WeakArrayKeyMap() 

361 self._temporary_memmaped_filenames = set() 

362 self._unlink_on_gc_collect = unlink_on_gc_collect 

363 

364 @property 

365 def _temp_folder(self): 

366 return self._temp_folder_resolver() 

367 

368 def __reduce__(self): 

369 # The ArrayMemmapForwardReducer is passed to the children processes: it 

370 # needs to be pickled but the _WeakArrayKeyMap need to be skipped as 

371 # it's only guaranteed to be consistent with the parent process memory 

372 # garbage collection. 

373 # Although this reducer is pickled, it is not needed in its destination 

374 # process (child processes), as we only use this reducer to send 

375 # memmaps from the parent process to the children processes. For this 

376 # reason, we can afford skipping the resolver, (which would otherwise 

377 # be unpicklable), and pass it as None instead. 

378 args = (self._max_nbytes, None, self._mmap_mode, 

379 self._unlink_on_gc_collect) 

380 kwargs = { 

381 'verbose': self.verbose, 

382 'prewarm': self._prewarm, 

383 } 

384 return ArrayMemmapForwardReducer, args, kwargs 

385 

386 def __call__(self, a): 

387 m = _get_backing_memmap(a) 

388 if m is not None and isinstance(m, np.memmap): 

389 # a is already backed by a memmap file, let's reuse it directly 

390 return _reduce_memmap_backed(a, m) 

391 

392 if (not a.dtype.hasobject and self._max_nbytes is not None and 

393 a.nbytes > self._max_nbytes): 

394 # check that the folder exists (lazily create the pool temp folder 

395 # if required) 

396 try: 

397 os.makedirs(self._temp_folder) 

398 os.chmod(self._temp_folder, FOLDER_PERMISSIONS) 

399 except OSError as e: 

400 if e.errno != errno.EEXIST: 

401 raise e 

402 

403 try: 

404 basename = self._memmaped_arrays.get(a) 

405 except KeyError: 

406 # Generate a new unique random filename. The process and thread 

407 # ids are only useful for debugging purpose and to make it 

408 # easier to cleanup orphaned files in case of hard process 

409 # kill (e.g. by "kill -9" or segfault). 

410 basename = "{}-{}-{}.pkl".format( 

411 os.getpid(), id(threading.current_thread()), uuid4().hex) 

412 self._memmaped_arrays.set(a, basename) 

413 filename = os.path.join(self._temp_folder, basename) 

414 

415 # In case the same array with the same content is passed several 

416 # times to the pool subprocess children, serialize it only once 

417 

418 is_new_memmap = filename not in self._temporary_memmaped_filenames 

419 

420 # add the memmap to the list of temporary memmaps created by joblib 

421 self._temporary_memmaped_filenames.add(filename) 

422 

423 if self._unlink_on_gc_collect: 

424 # Bump reference count of the memmap by 1 to account for 

425 # shared usage of the memmap by a child process. The 

426 # corresponding decref call will be executed upon calling 

427 # resource_tracker.maybe_unlink, registered as a finalizer in 

428 # the child. 

429 # the incref/decref calls here are only possible when the child 

430 # and the parent share the same resource_tracker. It is not the 

431 # case for the multiprocessing backend, but it does not matter 

432 # because unlinking a memmap from a child process is only 

433 # useful to control the memory usage of long-lasting child 

434 # processes, while the multiprocessing-based pools terminate 

435 # their workers at the end of a map() call. 

436 resource_tracker.register(filename, "file") 

437 

438 if is_new_memmap: 

439 # Incref each temporary memmap created by joblib one extra 

440 # time. This means that these memmaps will only be deleted 

441 # once an extra maybe_unlink() is called, which is done once 

442 # all the jobs have completed (or been canceled) in the 

443 # Parallel._terminate_backend() method. 

444 resource_tracker.register(filename, "file") 

445 

446 if not os.path.exists(filename): 

447 util.debug( 

448 "[ARRAY DUMP] Pickling new array (shape={}, dtype={}) " 

449 "creating a new memmap at {}".format( 

450 a.shape, a.dtype, filename)) 

451 for dumped_filename in dump(a, filename): 

452 os.chmod(dumped_filename, FILE_PERMISSIONS) 

453 

454 if self._prewarm: 

455 # Warm up the data by accessing it. This operation ensures 

456 # that the disk access required to create the memmapping 

457 # file are performed in the reducing process and avoids 

458 # concurrent memmap creation in multiple children 

459 # processes. 

460 load(filename, mmap_mode=self._mmap_mode).max() 

461 

462 else: 

463 util.debug( 

464 "[ARRAY DUMP] Pickling known array (shape={}, dtype={}) " 

465 "reusing memmap file: {}".format( 

466 a.shape, a.dtype, os.path.basename(filename))) 

467 

468 # The worker process will use joblib.load to memmap the data 

469 return ( 

470 (load_temporary_memmap, (filename, self._mmap_mode, 

471 self._unlink_on_gc_collect)) 

472 ) 

473 else: 

474 # do not convert a into memmap, let pickler do its usual copy with 

475 # the default system pickler 

476 util.debug( 

477 '[ARRAY DUMP] Pickling array (NO MEMMAPPING) (shape={}, ' 

478 ' dtype={}).'.format(a.shape, a.dtype)) 

479 return (loads, (dumps(a, protocol=HIGHEST_PROTOCOL),)) 

480 

481 

482def get_memmapping_reducers( 

483 forward_reducers=None, backward_reducers=None, 

484 temp_folder_resolver=None, max_nbytes=1e6, mmap_mode='r', verbose=0, 

485 prewarm=False, unlink_on_gc_collect=True, **kwargs): 

486 """Construct a pair of memmapping reducer linked to a tmpdir. 

487 

488 This function manage the creation and the clean up of the temporary folders 

489 underlying the memory maps and should be use to get the reducers necessary 

490 to construct joblib pool or executor. 

491 """ 

492 if forward_reducers is None: 

493 forward_reducers = dict() 

494 if backward_reducers is None: 

495 backward_reducers = dict() 

496 

497 if np is not None: 

498 # Register smart numpy.ndarray reducers that detects memmap backed 

499 # arrays and that is also able to dump to memmap large in-memory 

500 # arrays over the max_nbytes threshold 

501 forward_reduce_ndarray = ArrayMemmapForwardReducer( 

502 max_nbytes, temp_folder_resolver, mmap_mode, unlink_on_gc_collect, 

503 verbose, prewarm=prewarm) 

504 forward_reducers[np.ndarray] = forward_reduce_ndarray 

505 forward_reducers[np.memmap] = forward_reduce_ndarray 

506 

507 # Communication from child process to the parent process always 

508 # pickles in-memory numpy.ndarray without dumping them as memmap 

509 # to avoid confusing the caller and make it tricky to collect the 

510 # temporary folder 

511 backward_reducers[np.ndarray] = reduce_array_memmap_backward 

512 backward_reducers[np.memmap] = reduce_array_memmap_backward 

513 

514 return forward_reducers, backward_reducers 

515 

516 

517class TemporaryResourcesManager(object): 

518 """Stateful object able to manage temporary folder and pickles 

519 

520 It exposes: 

521 - a per-context folder name resolving API that memmap-based reducers will 

522 rely on to know where to pickle the temporary memmaps 

523 - a temporary file/folder management API that internally uses the 

524 resource_tracker. 

525 """ 

526 

527 def __init__(self, temp_folder_root=None, context_id=None): 

528 self._current_temp_folder = None 

529 self._temp_folder_root = temp_folder_root 

530 self._use_shared_mem = None 

531 self._cached_temp_folders = dict() 

532 self._id = uuid4().hex 

533 self._finalizers = {} 

534 if context_id is None: 

535 # It would be safer to not assign a default context id (less silent 

536 # bugs), but doing this while maintaining backward compatibility 

537 # with the previous, context-unaware version get_memmaping_executor 

538 # exposes too many low-level details. 

539 context_id = uuid4().hex 

540 self.set_current_context(context_id) 

541 

542 def set_current_context(self, context_id): 

543 self._current_context_id = context_id 

544 self.register_new_context(context_id) 

545 

546 def register_new_context(self, context_id): 

547 # Prepare a sub-folder name specific to a context (usually a unique id 

548 # generated by each instance of the Parallel class). Do not create in 

549 # advance to spare FS write access if no array is to be dumped). 

550 if context_id in self._cached_temp_folders: 

551 return 

552 else: 

553 # During its lifecycle, one Parallel object can have several 

554 # executors associated to it (for instance, if a loky worker raises 

555 # an exception, joblib shutdowns the executor and instantly 

556 # recreates a new one before raising the error - see 

557 # ``ensure_ready``. Because we don't want two executors tied to 

558 # the same Parallel object (and thus the same context id) to 

559 # register/use/delete the same folder, we also add an id specific 

560 # to the current Manager (and thus specific to its associated 

561 # executor) to the folder name. 

562 new_folder_name = ( 

563 "joblib_memmapping_folder_{}_{}_{}".format( 

564 os.getpid(), self._id, context_id) 

565 ) 

566 new_folder_path, _ = _get_temp_dir( 

567 new_folder_name, self._temp_folder_root 

568 ) 

569 self.register_folder_finalizer(new_folder_path, context_id) 

570 self._cached_temp_folders[context_id] = new_folder_path 

571 

572 def resolve_temp_folder_name(self): 

573 """Return a folder name specific to the currently activated context""" 

574 return self._cached_temp_folders[self._current_context_id] 

575 

576 # resource management API 

577 

578 def register_folder_finalizer(self, pool_subfolder, context_id): 

579 # Register the garbage collector at program exit in case caller forgets 

580 # to call terminate explicitly: note we do not pass any reference to 

581 # ensure that this callback won't prevent garbage collection of 

582 # parallel instance and related file handler resources such as POSIX 

583 # semaphores and pipes 

584 pool_module_name = whichmodule(delete_folder, 'delete_folder') 

585 resource_tracker.register(pool_subfolder, "folder") 

586 

587 def _cleanup(): 

588 # In some cases the Python runtime seems to set delete_folder to 

589 # None just before exiting when accessing the delete_folder 

590 # function from the closure namespace. So instead we reimport 

591 # the delete_folder function explicitly. 

592 # https://github.com/joblib/joblib/issues/328 

593 # We cannot just use from 'joblib.pool import delete_folder' 

594 # because joblib should only use relative imports to allow 

595 # easy vendoring. 

596 delete_folder = __import__( 

597 pool_module_name, fromlist=['delete_folder'] 

598 ).delete_folder 

599 try: 

600 delete_folder(pool_subfolder, allow_non_empty=True) 

601 resource_tracker.unregister(pool_subfolder, "folder") 

602 except OSError: 

603 warnings.warn("Failed to delete temporary folder: {}" 

604 .format(pool_subfolder)) 

605 

606 self._finalizers[context_id] = atexit.register(_cleanup) 

607 

608 def _clean_temporary_resources(self, context_id=None, force=False, 

609 allow_non_empty=False): 

610 """Clean temporary resources created by a process-based pool""" 

611 if context_id is None: 

612 # Iterates over a copy of the cache keys to avoid Error due to 

613 # iterating over a changing size dictionary. 

614 for context_id in list(self._cached_temp_folders): 

615 self._clean_temporary_resources( 

616 context_id, force=force, allow_non_empty=allow_non_empty 

617 ) 

618 else: 

619 temp_folder = self._cached_temp_folders.get(context_id) 

620 if temp_folder and os.path.exists(temp_folder): 

621 for filename in os.listdir(temp_folder): 

622 if force: 

623 # Some workers have failed and the ref counted might 

624 # be off. The workers should have shut down by this 

625 # time so forcefully clean up the files. 

626 resource_tracker.unregister( 

627 os.path.join(temp_folder, filename), "file" 

628 ) 

629 else: 

630 resource_tracker.maybe_unlink( 

631 os.path.join(temp_folder, filename), "file" 

632 ) 

633 

634 # When forcing clean-up, try to delete the folder even if some 

635 # files are still in it. Otherwise, try to delete the folder 

636 allow_non_empty |= force 

637 

638 # Clean up the folder if possible, either if it is empty or 

639 # if none of the files in it are in used and allow_non_empty. 

640 try: 

641 delete_folder( 

642 temp_folder, allow_non_empty=allow_non_empty 

643 ) 

644 # Forget the folder once it has been deleted 

645 self._cached_temp_folders.pop(context_id, None) 

646 resource_tracker.unregister(temp_folder, "folder") 

647 

648 # Also cancel the finalizers that gets triggered at gc. 

649 finalizer = self._finalizers.pop(context_id, None) 

650 if finalizer is not None: 

651 atexit.unregister(finalizer) 

652 

653 except OSError: 

654 # Temporary folder cannot be deleted right now. 

655 # This folder will be cleaned up by an atexit 

656 # finalizer registered by the memmapping_reducer. 

657 pass