Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/joblib/disk.py: 28%

1"""

2Disk management utilities.

3"""

5# Authors: Gael Varoquaux <gael dot varoquaux at normalesup dot org>

6# Lars Buitinck

8# License: BSD Style, 3 clauses.

10import errno

11import os

12import shutil

13import sys

14import time

15from multiprocessing import util

17try:

18 WindowsError

19except NameError:

20 WindowsError = OSError

23def disk_used(path):

24 """Return the disk usage in a directory."""

25 size = 0

26 for file in os.listdir(path) + ["."]:

27 stat = os.stat(os.path.join(path, file))

28 if hasattr(stat, "st_blocks"):

29 size += stat.st_blocks * 512

30 else:

31 # on some platform st_blocks is not available (e.g., Windows)

32 # approximate by rounding to next multiple of 512

33 size += (stat.st_size // 512 + 1) * 512

34 # We need to convert to int to avoid having longs on some systems (we

35 # don't want longs to avoid problems we SQLite)

36 return int(size / 1024.0)

39def memstr_to_bytes(text):

40 """Convert a memory text to its value in bytes."""

41 kilo = 1024

42 units = dict(K=kilo, M=kilo**2, G=kilo**3)

43 try:

44 size = int(units[text[-1]] * float(text[:-1]))

45 except (KeyError, ValueError) as e:

46 raise ValueError(

47 "Invalid literal for size give: %s (type %s) should be "

48 "alike '10G', '500M', '50K'." % (text, type(text))

49 ) from e

50 return size

53def mkdirp(d):

54 """Ensure directory d exists (like mkdir -p on Unix)

55 No guarantee that the directory is writable.

56 """

57 try:

58 os.makedirs(d)

59 except OSError as e:

60 if e.errno != errno.EEXIST:

61 raise

64# if a rmtree operation fails in rm_subdirs, wait for this much time (in secs),

65# then retry up to RM_SUBDIRS_N_RETRY times. If it still fails, raise the

66# exception. this mechanism ensures that the sub-process gc have the time to

67# collect and close the memmaps before we fail.

68RM_SUBDIRS_RETRY_TIME = 0.1

69RM_SUBDIRS_N_RETRY = 10

72def rm_subdirs(path, onerror=None):

73 """Remove all subdirectories in this path.

75 The directory indicated by `path` is left in place, and its subdirectories

76 are erased.

78 If onerror is set, it is called to handle the error with arguments (func,

79 path, exc_info) where func is os.listdir, os.remove, or os.rmdir;

80 path is the argument to that function that caused it to fail; and

81 exc_info is a tuple returned by sys.exc_info(). If onerror is None,

82 an exception is raised.

83 """

85 # NOTE this code is adapted from the one in shutil.rmtree, and is

86 # just as fast

88 names = []

89 try:

90 names = os.listdir(path)

91 except os.error:

92 if onerror is not None:

93 onerror(os.listdir, path, sys.exc_info())

94 else:

95 raise

97 for name in names:

98 fullname = os.path.join(path, name)

99 delete_folder(fullname, onerror=onerror)

100

101

102def delete_folder(folder_path, onerror=None, allow_non_empty=True):

103 """Utility function to cleanup a temporary folder if it still exists."""

104 if os.path.isdir(folder_path):

105 if onerror is not None:

106 shutil.rmtree(folder_path, False, onerror)

107 else:

108 # allow the rmtree to fail once, wait and re-try.

109 # if the error is raised again, fail

110 err_count = 0

111 while True:

112 files = os.listdir(folder_path)

113 try:

114 if len(files) == 0 or allow_non_empty:

115 shutil.rmtree(folder_path, ignore_errors=False, onerror=None)

116 util.debug("Successfully deleted {}".format(folder_path))

117 break

118 else:

119 raise OSError(

120 "Expected empty folder {} but got {} files.".format(

121 folder_path, len(files)

122 )

123 )

124 except (OSError, WindowsError):

125 err_count += 1

126 if err_count > RM_SUBDIRS_N_RETRY:

127 # the folder cannot be deleted right now. It maybe

128 # because some temporary files have not been deleted

129 # yet.

130 raise

131 time.sleep(RM_SUBDIRS_RETRY_TIME)