i'm still trying to get myself to engage this nettensors code more ... it's
hard to stay with
there was idea that likely those rotor motors are too weak if not geared, this
is quite possible, i don't know
worried i'll make this nettensors code even messier
here's the total current function blob. this is not a cohesive function that
makes sense, it's colelctions of my amnesiatic attempts to implement it from
copypasting a different one in.
def read_many(self, offset_lengths, progress, validate_sorted=True):
if validate_sorted:
sorted_offset_lengths = list(offset_lengths)
sorted_offset_lengths.sort()
assert sorted_offset_lengths == offset_lengths
OP_FETCH = 1
OP_PLACE = 2
OP_OUTPUT = 4
offset_length_tail_idx_ops =
torch.zeros([offset_lengths.shape[0]*2, 5])
OFFSET, LENGTH, TAIL, IDX, OP =
range(offset_length_tail_ops.shape[-1])
op_ct = 0
#results = torch.empty(len(o
#results = [None] * len(offset_lengths)
tails = (offset_lengths[:,0] +
offset_lengths[:,1]).clamp(max=len(self.mmap))
aligned_offsets = offset_lengths[:,0] // self.blksize;
aligned_offsets *= self.blksize
aligned_tails = (tails - 1); aligned_tails //= self.blksize;
aligned_tails += 1; aligned_tails *= self.blksize; torch.clamp(aligned_tails,
max=self.size(), out=aligned_tails)
cls = type(self.fetchers)
avail_disk_space = (psutil.disk_usage(self.fn).free +
cls.sparse_usage) * self.fetchers.usage_frac - cls.sparse_usage
min_hole = 0
pbar = range(len(offset_lengths))
if progress:
pbar = tqdm.tqdm(pbar, total=len(offset_lengths),
desc=progress, leave=False, unit='rd')
idx = 0
while idx < len(offset_lengths):
#for idx in pbar:#range(len(offset_lengths)):
#offset, length = offset_lengths[idx]
#tail = min(offset + length, len(self.mmap))
#aligned_offset = (offset // self.blksize) * self.blksize
#aligned_tail = min(self.size(), (((tail - 1) // self.blksize)
+ 1) * self.blksize)
aligned_offset = aligned_offsets[idx].item()
next_hole = self._next_sparse(max(aligned_offset, min_hole),
os.SEEK_HOLE)
# 1/3: COMPLETELY CACHED ITEMS this description of the subset
looks correct
cached_idcs = (tails[idx:] < next_hole).nonzero()[:,0]
num_cached_idcs = cached_idcs.shape[0]
if num_cached_idcs > 0:
next_idx = idx + num_cached_idcs
assert (cached_idcs < next_idx).all()
next_op_ct = op_ct + num_cached_idcs
offset_length_tail_ops[op_ct:next_op_ct,[OFFSET,LENGTH]] =
offset_lengths[idx:next_idx]
offset_length_tail_ops[op_ct:next_op_ct,TAIL] =
tails[idx:next_idx]
offset_length_tail_ops[op_ct:next_op_ct,IDX] = cached_idcs
offset_length_tail_ops[op_ct:next_op_ct,OP] = OP_OUTPUT
op_ct = next_op_ct
idx = next_idx
next_data = self._next_sparse(next_hole, os.SEEK_DATA)
# 2/3: COMPLETELY UNCACHED ITEMS
missing_idcs = (tails[idx:] < next_data).nonzero()[:,0]
#missing_idcs = (next_hole < tails[idx:] and next_data >
offsets_lengths[idx:,0]).nonzero()[:,0]
#missing_idcs = (next_hole < tails[idx:]).nonzero()[:,0]
# here we are handling all undownloaded indices before the next
cached ones
# there could be many pages between them that don't need to be
fetched
num_missing_idcs = missing_idcs.shape[0]
if num_missing_idcs > 0:
# uncached data
next_idx = idx + num_missing_idcs
assert (missing_idcs < next_idx).all() but most likely
there could be some idcs that are not missing ... ones where tail < next_hole?
missing_offset_lengths = offset_lengths[idx:next_idx]
if missing_offset_lengths[:,1].sum() > avail_disk_space:
# note: SMALL BUG here in that it checks the tensor
size instead of the surrounding pages actually fetched
# no more disk space
if not cls.warned_space:
import warnings
warnings.warn(
'\nCACHE FULL CACHE FULL' +
'\nRequested=' +
str(tqdm.tqdm.format_sizeof(aligned_tail -
aligned_offset, 'B', 1024))
+ ' Cached=' +
str(tqdm.tqdm.format_sizeof(cls.sparse_usage,
'B', 1024))
+ ' Free=' +
str(tqdm.tqdm.format_sizeof(psutil.disk_usage(self.fn).free, 'B', 1024))
+ '\n' +
os.path.dirname(self.fn)
+ '\nCACHE FULL CACHE FULL'
, stacklevel=5)
cls.warned_space = True
next_op_ct = op_ct + num_missing_idcs
offset_length_tail_ops[op_ct:next_op_ct,[OFFSET,LENGTH]] =
missing_offset_lengths
offset_length_tail_ops[op_ct:next_op_ct,TAIL] =
tails[idx:next_idx]
offset_length_tail_ops[op_ct:next_op_ct,IDX] =
missing_idcs
offset_length_tail_ops[op_ct:next_op_ct,OP] = OP_FETCH
| OP_OUTPUT
op_ct = next_op_ct
idx = next_idx
continue
# now we want to group these into those that do not have an
empty page between them, and we can fetch everything underneath them at once
g # note that there may be a tensor that is partly filled,
and may have further holes farther along, at the right edge
#if num_missing_idcs > 1:
if True:
# add fetches, placements, and outputs covering all
pages with tensors wholly in them
# each tensor has an aligned offset and aligned
tail already
# it might be more useful to consider abstract
alignment points that information would be rounded too
# we could calculate empty regions from the
differences between the offsets and tails
# empty regions that don't contain pages would be
elided away
# empty regions that do contain pages would be
aligned
# alternatively one could compare aligned_offsets
and aligned_tails for equality and overlap
aligned_start = aligned_offsets[idx]
aligned_end = min(aligned_tails[next_idx-1], next_hole)
#empty_regions = missing_offset_lengths[:-1, 0]
assert not (tails[idx:next_idx-1] >
offsets[idx+1:next_idx]).any()
#mergeable_regions = aligned_tails[idx:next_idx-1] >=
aligned_offsets[idx+1:next_idx]
#merge_bounds = mergeable_regions[:-1] !=
mergeable_regions[1:]
# so here first we compare the preceding tails to the
following offsets
# mergeable_regions then provides a bool for each space
between regions that represents the two being the same. so there's 1 less bool
and it's the spaces between regions.
# merge_bounds then looks for these regions of sameness
within mergeable_regions. so merge_bounds has a length of 2 fewer than
originally, and it relates to comparisons between adjacent pairs of spaces
between regions.
# the data starts and ends with useful groups .. but
this may not yet be represented in merge_bounds
# might want to special case length==2 somewhere
for ease
# it's notable that mergeable_regions[x]
indicates if True that the region can be merged with its neighbors
# but that if False it indicates that the
region needs to be treated as its own single group
region_mask = aligned_tails[idx:next_idx-1] <
aligned_offsets[idx+1:next_idx]
region_bounds = region_mask.nonzero()[:,0]
region_bounds += idx
# region bounds says which offsets+1 cannot
be replaced by their preceding offsets (if idx shifted by 1)
# that is, which tails cannot be replaced
by their following tails
# T T F F T F T true if the tail at that index
is less than the offst of the following -- ie a gap follows
# we want head + tail from 0 and 1
because a gap follows them
# then we want head from 2, and tail from
4
# then head from 5, and tail from 6
# 0 1 2 3 4 5 6
# 0, 1, 4, 6
# head and tail from: -1+1=0, 0
# 0+1=1, 1
# 1+1=2, 4
# 4+1=5, 6
# note these are double indices
next_op_ct = op_ct + region_bounds.shape[0] + 1
offset_length_tail_idx_ops[op_ct:next_op_ct-1,TAIL] =
aligned_tails[region_bounds]
offset_length_tail_idx_ops[next_op_ct-1,TAIL] =
aligned_end
# why is the offset of one, one more than the
tail of the preceding one
# maybe an indexing operation not performed?
#offset_length_tail_idx_ops[op_ct+1:next_op_ct,OFFSET]
= offset_length_tail_idx_ops[op_ct:next_op_ct-1,TAIL]
#offset_length_tail_idx_ops[op_ct+1:next_op_ct,OFFSET]
+= 1
region_bounds += 1
offset_length_tail_idx_ops[op_ct+1:next_op_ct,OFFSET] =
aligned_offsets[region_bounds]
offset_length_tail_idx_ops[op_ct,OFFSET] = aligned_start
offset_length_tail_idx_ops[op_ct:next_op_ct,OP] =
OP_FETCH | OP_PLACE
#if mergeable_regions[0]:
# # first pair is mergeable
# # merge_bouds[0] represents the end of the first
merge
#else:
# # first pair is not mergeable
# # merge_bounds[0] represents the start of the
first merge
#merged_aligned_offsets = aligned_offsets[
## how to now merge them; i guess extract the start and
end
# what remains after all the merged placements above is
the last one
# it may not need to be fully fetched
# ... it calculates a bool regarding whether the last
tail is distinct from the second-to-last
# but there are only n-1 calcs ... one is added at the
start with the initial offset ...
g # consider [oop
#
#torch.minimum(offset_length_tail_idx_ops[next_op_ct-1,TAIL], next_hole)
##### now under the condition of num_missing_idcs > 1,
# now,
# ops have hopefully been set to place everything up
to next_hole
# it may be such that tail[next_idx-1] > next_hole in
which case more must be done
we can maybe do hole on left for every placement
#if aligned_offset - 1 >= min_hole:
# hole_on_left = self._next_sparse(aligned_offset - 1,
os.SEEK_HOLE) < aligned_offset
#else:
# hole_on_left = False
#
# hole_on_left = self._next_sparse(
# 3/3: 1 PARTLY CACHED ITEM, possibly with multiple
scattered holes inside 1 item
#next_data = self._next_sparse(next_hole, os.SEEK_DATA)
tail = tails[next_idx-1]
aligned_tail = aligned_tails[next_idx-1]
aligned_offset =
offset_length_tail_idx_ops[next_op_ct-1,TAIL]
length = aligned_tail - aligned_offset
while next_data < tail:
assert next_data - next_hole <= length
length = next_data - next_hole
offset_length_tail_idx_ops[next_op_ct,OFFSET] =
next_hole
offset_length_tail_idx_ops[next_op_ct,TAIL] = next_data
offset_length_tail_idx_ops[next_op_ct,OP] = OP_FETCH |
OP_PLACE
next_op_ct += 1
cls.sparse_usage += length
next_hole = self._next_sparse(next_data, os.SEEK_HOLE)
next_data = self._next_sparse(next_hole, os.SEEK_DATA)
if next_hole < tail:
length = aligned_tail - next_hole
offset_length_tail_idx_ops[next_op_ct,OFFSET] =
next_hole
offset_length_tail_idx_ops[next_op_ct,TAIL] =
aligned_tail
offset_length_tail_idx_ops[next_op_ct,OP] = OP_FETCH |
OP_PLACE
next_op_ct += 1
cls.sparse_usage += length
next_hole = self._next_sparse(aligned_tail,
os.SEEK_HOLE)
offset_length_tail_idx_ops[op_ct:next_op_ct,LENGTH] =
offset_length_tail_idx_ops[op_ct:next_op_ct,TAIL]
offset_length_tail_idx_ops[op_ct:next_op_ct,LENGTH] -=
offset_length_tail_idx_ops[op_ct:next_op_ct,OFFSET]
cls.sparse_usage +=
offset_length_tail_idx_ops[op_ct:next_op_ct,LENGTH].sum()
op_ct = next_op_ct
min_hole = max(next_hole, min_hole)
next_op_ct = op_ct +
offset_length_tail_idx_ops
if next_hole < tail:
# data not cached
if cls.sparse_usage + aligned_tail - aligned_offset >
(psutil.disk_usage(self.fn).free + cls.sparse_usage) * self.fetchers.usage_frac:
# no more disk space
if not cls.warned_space:
import warnings
warnings.warn(
'\nCACHE FULL CACHE FULL' +
'\nRequested=' +
str(tqdm.tqdm.format_sizeof(aligned_tail -
aligned_offset, 'B', 1024))
+ ' Cached=' +
str(tqdm.tqdm.format_sizeof(cls.sparse_usage,
'B', 1024))
+ ' Free=' +
str(tqdm.tqdm.format_sizeof(psutil.disk_usage(self.fn).free, 'B', 1024))
+ '\n' +
os.path.dirname(self.fn)
+ '\nCACHE FULL CACHE FULL'
, stacklevel=5)
cls.warned_space = True
fetch_outputs.append([len(fetches), idx])
fetches.append([offset, length])
continue
#return super().read(offset, length, progress=progress)
hole_on_left = self._next_sparse(max(aligned_offset - 1,
min_hole), os.SEEK_HOLE) < aligned_offset
length = aligned_tail - aligned_offset
next_data = self._next_sparse(next_hole, os.SEEK_DATA)
while next_data < tail:
assert next_data - next_hole <= length
length = next_data - next_hole
placements.append([len(fetches), next_hole, next_data])
fetches.append([next_hole, length])
#self.mmap[next_hole:next_data] =
super().read(next_hole, length, progress=progress)
cls.sparse_usage += length
next_hole = self._next_sparse(next_data, os.SEEK_HOLE)
next_data = self._next_sparse(next_hole, os.SEEK_DATA)
if next_hole < tail:
length = aligned_tail - next_hole
placements.append([len(fetches), next_hole,
aligned_tail])
fetches.append([next_hole, length])
#self.mmap[next_hole:aligned_tail] =
super().read(next_hole, length, progress=progress)
cls.sparse_usage += length
# updated this while sleepy
# on docker vms i found the memory mapper filling extra
blocks with 0s
# this new code tries to ensure data is correct when
that happens
# i've also updated the pagesize calculation so this
might happen less
next_hole = self._next_sparse(aligned_tail,
os.SEEK_HOLE)
extra_0s_right = min(next_hole, next_data)
while extra_0s_right > aligned_tail:
length = extra_0s_right - aligned_tail
placements.append([len(fetches), aligned_tail,
extra_0s_right])
fetches.append([aligned_tail, length])
#self.mmap[aligned_tail:extra_0s_right] =
super().read(aligned_tail, length, progress=progress)
cls.sparse_usage += length
next_hole = self._next_sparse(extra_0s_right,
os.SEEK_HOLE)
extra_0s_right = min(next_hole, next_data)
min_hole = max(next_hole, min_hole)
if hole_on_left:
check_holes_on_left.append(aligned_offset)
# if self._next_sparse(aligned_offset - 1, os.SEEK_HOLE)
>= aligned_offset:
place_outputs.append([offset, tail, idx])
#return self.mmap[offset:tail]
if progress:
pbar.close()
if len(fetches):
fetches = super().read_many(fetches, progress=progress,
validate_sorted=False)
for fetchidx, start, end in placements:
self.mmap[start:end] = fetches[fetchidx]
for fetchidx, resultidx in fetch_outputs:
results[resultidx] = fetches[fetchidx]
for start, end, resultidx in place_outputs:
results[resultidx] = self.mmap[offset:tail]
for check_hole in check_holes_on_left:
if self._next_sparse(check_hole - 1, os.SEEK_HOLE) >=
check_hole:
# a hole on the left disappeared
# this could be resolved by walking holes on the left or
storing auxiliary data regarding allocated regions
# the former is space efficient and the latter time
efficient; they could be combined as well
os.unlink(self.fn)
raise Exception(
'Your memory mapper is writing data below the cached
region ' +
'even when aligned to the pagesize and blocksize. ' +
'The current code generates corrupt cached runs of 0s
in this situation.')
return results
