File size: 2,547 Bytes
4bcf80e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 | """Inspect page residency and control only a task-owned checkpoint copy."""
import ctypes
import os
from pathlib import Path
import time
def residency(path):
page = os.sysconf('SC_PAGE_SIZE')
size = path.stat().st_size
libc = ctypes.CDLL(None, use_errno=True)
libc.mmap.restype = ctypes.c_void_p
libc.mmap.argtypes = [ctypes.c_void_p,ctypes.c_size_t,ctypes.c_int,ctypes.c_int,ctypes.c_int,ctypes.c_long]
libc.mincore.argtypes = [ctypes.c_void_p,ctypes.c_size_t,ctypes.c_void_p]
libc.munmap.argtypes = [ctypes.c_void_p,ctypes.c_size_t]
with path.open('rb') as stream:
address = libc.mmap(None,size,0,1,stream.fileno(),0)
if address == ctypes.c_void_p(-1).value:
raise OSError(ctypes.get_errno(),'mmap failed')
try:
pages = (size+page-1)//page
values = (ctypes.c_ubyte*pages)()
if libc.mincore(address,size,values):
raise OSError(ctypes.get_errno(),'mincore failed')
return dict(bytes=size,pages=pages,resident_pages=sum(bool(v&1) for v in values))
finally:
libc.munmap(address,size)
def prepare(snapshot,mode,task_root):
snapshot = Path(snapshot).resolve()
files = sorted(snapshot.glob('*.safetensors'))
started = time.perf_counter()
if mode == 'cold':
private = (Path(task_root)/'isolated-checkpoints').resolve()
if not snapshot.is_relative_to(private) or any(not file.resolve().is_relative_to(private) for file in files):
raise ValueError('Cold-cache control requires a private, nonsymlinked checkpoint copy')
for file in files:
with file.open('rb') as stream:
os.fsync(stream.fileno())
os.posix_fadvise(stream.fileno(),0,0,os.POSIX_FADV_DONTNEED)
elif mode == 'warm':
buffer = bytearray(8*1024**2)
for file in files:
with file.open('rb',buffering=0) as stream:
while stream.readinto(buffer):
pass
else:
raise ValueError('Choose cold or warm preparation')
status = {file.name:residency(file) for file in files}
if mode == 'cold' and any(v['resident_pages']/v['pages']>.01 for v in status.values()):
raise RuntimeError('Private checkpoint did not become cold; do not label this a cold-storage trial')
return dict(mode=mode,preparation_seconds=time.perf_counter()-started,files=status,
boundary='Page residency immediately before container launch; no system-wide cache dropping')
|