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)
69 # only set the event when mmap is successful
72 if tmpfile is not None:
73 # If the tempfile hasn't been renamed on disk yet, try to delete it.
80 if self.content is None:
83 return len(self.content)
86 if self.content is not None and len(self.content) > 0:
87 # The mmap region might be in use when we decided to evict
88 # it. This can happen if the cache is too small.
90 # If we call close() now, it'll throw an error if
91 # something tries to access it.
93 # However, we don't need to explicitly call mmap.close()
95 # I confirmed in mmapmodule.c that that both close
96 # and deallocate do the same thing:
98 # a) close the file descriptor
99 # b) unmap the memory range
101 # So we can forget it in the cache and delete the file on
102 # disk, and it will tear it down after any other
103 # lingering Python references to the mapped memory are
106 blockdir = os.path.join(self.cachedir, self.locator[0:3])
107 final = os.path.join(blockdir, self.locator) + cacheblock_suffix
109 fcntl.flock(self.filehandle, fcntl.LOCK_UN)
111 # try to get an exclusive lock, this ensures other
112 # processes are not using the block. It is
113 # nonblocking and will throw an exception if we
114 # can't get it, which is fine because that means
115 # we just won't try to delete it.
117 # I should note here, the file locking is not
118 # strictly necessary, we could just remove it and
119 # the kernel would ensure that the underlying
120 # inode remains available as long as other
121 # processes still have the file open. However, if
122 # you have multiple processes sharing the cache
123 # and deleting each other's files, you'll end up
124 # with a bunch of ghost files that don't show up
125 # in the file system but are still taking up
126 # space, which isn't particularly user friendly.
127 # The locking strategy ensures that cache blocks
128 # in use remain visible.
130 fcntl.flock(self.filehandle, fcntl.LOCK_EX | fcntl.LOCK_NB)
137 self.filehandle = None
142 def get_from_disk(locator, cachedir):
143 blockdir = os.path.join(cachedir, locator[0:3])
144 final = os.path.join(blockdir, locator) + cacheblock_suffix
147 filehandle = open(final, "rb")
149 # aquire a shared lock, this tells other processes that
150 # we're using this block and to please not delete it.
151 fcntl.flock(filehandle, fcntl.LOCK_SH)
153 content = mmap.mmap(filehandle.fileno(), 0, access=mmap.ACCESS_READ)
154 dc = DiskCacheSlot(locator, cachedir)
155 dc.filehandle = filehandle
159 except FileNotFoundError:
161 except Exception as e:
162 traceback.print_exc()
167 def cache_usage(cachedir):
169 for root, dirs, files in os.walk(cachedir):
171 if not name.endswith(cacheblock_suffix):
174 blockpath = os.path.join(root, name)
175 res = os.stat(blockpath)
181 def init_cache(cachedir, maxslots):
183 # First check the disk cache works at all by creating a 1 byte cache entry
185 checkexists = DiskCacheSlot.get_from_disk('0cc175b9c0f1b6a831c399e269772661', cachedir)
186 ds = DiskCacheSlot('0cc175b9c0f1b6a831c399e269772661', cachedir)
188 if checkexists is None:
189 # Don't keep the test entry around unless it existed beforehand.
192 # map in all the files in the cache directory, up to max slots.
193 # after max slots, try to delete the excess blocks.
195 # this gives the calling process ownership of all the blocks
198 for root, dirs, files in os.walk(cachedir):
200 if not name.endswith(cacheblock_suffix):
203 blockpath = os.path.join(root, name)
204 res = os.stat(blockpath)
206 if len(name) == (32+len(cacheblock_suffix)) and not name.startswith("tmp"):
207 blocks.append((name[0:32], res.st_atime))
208 elif name.startswith("tmp") and ((time.time() - res.st_mtime) > 60):
209 # found a temporary file more than 1 minute old,
216 # sort by access time (atime), going from most recently
217 # accessed (highest timestamp) to least recently accessed
218 # (lowest timestamp).
219 blocks.sort(key=lambda x: x[1], reverse=True)
221 # Map in all the files we found, up to maxslots, if we exceed
222 # maxslots, start throwing things out.
225 got = DiskCacheSlot.get_from_disk(b[0], cachedir)
228 if len(cachelist) < maxslots:
229 cachelist.append(got)
231 # we found more blocks than maxslots, try to
232 # throw it out of the cache.