1 # Copyright (C) The Arvados Authors. All rights reserved.
3 # SPDX-License-Identifier: Apache-2.0
16 _logger = logging.getLogger('arvados.keep')
18 cacheblock_suffix = ".keepcacheblock"
20 class DiskCacheSlot(object):
21 __slots__ = ("locator", "ready", "content", "cachedir", "filehandle")
23 def __init__(self, locator, cachedir):
24 self.locator = locator
25 self.ready = threading.Event()
27 self.cachedir = cachedir
28 self.filehandle = None
42 # Can't mmap a 0 length file
46 if self.content is not None:
47 # Has been set already
50 blockdir = os.path.join(self.cachedir, self.locator[0:3])
51 os.makedirs(blockdir, mode=0o700, exist_ok=True)
53 final = os.path.join(blockdir, self.locator) + cacheblock_suffix
55 self.filehandle = tempfile.NamedTemporaryFile(dir=blockdir, delete=False, prefix="tmp", suffix=cacheblock_suffix)
56 tmpfile = self.filehandle.name
57 os.chmod(tmpfile, stat.S_IRUSR | stat.S_IWUSR)
59 # aquire a shared lock, this tells other processes that
60 # we're using this block and to please not delete it.
61 fcntl.flock(self.filehandle, fcntl.LOCK_SH)
63 self.filehandle.write(value)
64 self.filehandle.flush()
65 os.rename(tmpfile, final)
68 self.content = mmap.mmap(self.filehandle.fileno(), 0, access=mmap.ACCESS_READ)
70 if tmpfile is not None:
71 # If the tempfile hasn't been renamed on disk yet, try to delete it.
78 if self.content is None:
81 return len(self.content)
84 if self.content is not None and len(self.content) > 0:
85 # The mmap region might be in use when we decided to evict
86 # it. This can happen if the cache is too small.
88 # If we call close() now, it'll throw an error if
89 # something tries to access it.
91 # However, we don't need to explicitly call mmap.close()
93 # I confirmed in mmapmodule.c that that both close
94 # and deallocate do the same thing:
96 # a) close the file descriptor
97 # b) unmap the memory range
99 # So we can forget it in the cache and delete the file on
100 # disk, and it will tear it down after any other
101 # lingering Python references to the mapped memory are
104 blockdir = os.path.join(self.cachedir, self.locator[0:3])
105 final = os.path.join(blockdir, self.locator) + cacheblock_suffix
107 fcntl.flock(self.filehandle, fcntl.LOCK_UN)
109 # try to get an exclusive lock, this ensures other
110 # processes are not using the block. It is
111 # nonblocking and will throw an exception if we
112 # can't get it, which is fine because that means
113 # we just won't try to delete it.
115 # I should note here, the file locking is not
116 # strictly necessary, we could just remove it and
117 # the kernel would ensure that the underlying
118 # inode remains available as long as other
119 # processes still have the file open. However, if
120 # you have multiple processes sharing the cache
121 # and deleting each other's files, you'll end up
122 # with a bunch of ghost files that don't show up
123 # in the file system but are still taking up
124 # space, which isn't particularly user friendly.
125 # The locking strategy ensures that cache blocks
126 # in use remain visible.
128 fcntl.flock(self.filehandle, fcntl.LOCK_EX | fcntl.LOCK_NB)
135 self.filehandle = None
140 def get_from_disk(locator, cachedir):
141 blockdir = os.path.join(cachedir, locator[0:3])
142 final = os.path.join(blockdir, locator) + cacheblock_suffix
145 filehandle = open(final, "rb")
147 # aquire a shared lock, this tells other processes that
148 # we're using this block and to please not delete it.
149 fcntl.flock(filehandle, fcntl.LOCK_SH)
151 content = mmap.mmap(filehandle.fileno(), 0, access=mmap.ACCESS_READ)
152 dc = DiskCacheSlot(locator, cachedir)
153 dc.filehandle = filehandle
157 except FileNotFoundError:
159 except Exception as e:
160 traceback.print_exc()
165 def cache_usage(cachedir):
167 for root, dirs, files in os.walk(cachedir):
169 if not name.endswith(cacheblock_suffix):
172 blockpath = os.path.join(root, name)
173 res = os.stat(blockpath)
179 def init_cache(cachedir, maxslots):
181 # First check the disk cache works at all by creating a 1 byte cache entry
183 checkexists = DiskCacheSlot.get_from_disk('0cc175b9c0f1b6a831c399e269772661', cachedir)
184 ds = DiskCacheSlot('0cc175b9c0f1b6a831c399e269772661', cachedir)
186 if checkexists is None:
187 # Don't keep the test entry around unless it existed beforehand.
190 # map in all the files in the cache directory, up to max slots.
191 # after max slots, try to delete the excess blocks.
193 # this gives the calling process ownership of all the blocks
196 for root, dirs, files in os.walk(cachedir):
198 if not name.endswith(cacheblock_suffix):
201 blockpath = os.path.join(root, name)
202 res = os.stat(blockpath)
204 if len(name) == (32+len(cacheblock_suffix)) and not name.startswith("tmp"):
205 blocks.append((name[0:32], res.st_atime))
206 elif name.startswith("tmp") and ((time.time() - res.st_mtime) > 60):
207 # found a temporary file more than 1 minute old,
214 # sort by access time (atime), going from most recently
215 # accessed (highest timestamp) to least recently accessed
216 # (lowest timestamp).
217 blocks.sort(key=lambda x: x[1], reverse=True)
219 # Map in all the files we found, up to maxslots, if we exceed
220 # maxslots, start throwing things out.
223 got = DiskCacheSlot.get_from_disk(b[0], cachedir)
226 if len(cachelist) < maxslots:
227 cachelist.append(got)
229 # we found more blocks than maxslots, try to
230 # throw it out of the cache.