Add SharedBytes.copyToCacheFileAligned without length method #106193
Conversation
tlrx
added
>non-issue
:Distributed Indexing/Engine
|
Pinging @elastic/es-distributed (Team:Distributed) |
elasticsearchmachine
added
the
Team:Distributed (Obsolete)
henningandersen
left a comment
henningandersen
left a comment
There was a problem hiding this comment.
LGTM.
Ideally we'd add a test here for the new method, but I can see how that might be cumbersome. If that is the reasoning I am ok to proceed as is, we'll test this from stateless then.
| final RangeAvailableHandler reader, | ||
| final RangeMissingHandler writer | ||
| ) throws Exception { | ||
| assert assertOffsetsWithinFileLength(rangeToWrite.start(), rangeToWrite.length(), length); |
There was a problem hiding this comment.
Would be good to add assert rangeToWrite.start() >= 0 instead.
There was a problem hiding this comment.
Might also be great to comment on why we allow rangeToWrite beyond the blob length since it is not intuitive.
| } | ||
|
|
||
| public static ByteRange computeRange(long rangeSize, long position, long size) { | ||
| return ByteRange.of((position / rangeSize) * rangeSize, (((position + size - 1) / rangeSize) + 1) * rangeSize); |
There was a problem hiding this comment.
Does it make sense to reuse computeRange(rangeSize, position, size, Long.MAX_VALUE) instead?
There was a problem hiding this comment.
I suppose it would be OK, but I prefer to have a dedicated method.
| final RangeAvailableHandler reader, | ||
| final RangeMissingHandler writer | ||
| ) throws Exception { | ||
| assert assertOffsetsWithinFileLength(rangeToWrite.start(), rangeToWrite.length(), length); |
There was a problem hiding this comment.
Might also be great to comment on why we allow rangeToWrite beyond the blob length since it is not intuitive.
| if (buffer.hasRemaining()) { | ||
| // ensure that last write is aligned on 4k boundaries (= page size) | ||
| final int remainder = buffer.position() % PAGE_SIZE; | ||
| final int adjustment = remainder == 0 ? 0 : PAGE_SIZE - remainder; | ||
| buffer.position(buffer.position() + adjustment); | ||
| } | ||
| bytesCopied += positionalWrite(fc, fileChannelPos + bytesCopied, buffer); | ||
| progressUpdater.accept(bytesCopied); |
There was a problem hiding this comment.
Mostly for my education: With the existing version of this method, we adjust bytesCopied with the page alignment before invoke progressUpdater. I think this is beacuse ProgressListenableActionFuture has a few assertions that expect bytesCopied not exceeding its end value (example).
We don't need the same adjustment here for the callback because ProgressListenableActionFuture should always be created with page aligned end value and this is guaranteed by the new BlobCacheUtils#computeRange method?
There was a problem hiding this comment.
I think this is beacuse
ProgressListenableActionFuturehas a few assertions that expectbytesCopiednot exceeding itsendvalue (example).
Yes, we cannot update progress to a value that would exceed the pending range in the SparseFileTracker.
We don't need the same adjustment here for the callback because
ProgressListenableActionFutureshould always be created with page alignedendvalue and this is guaranteed by the newBlobCacheUtils#computeRangemethod?
Yes 👍
|
Thanks everyone! |
5 participants
This change adds a variant of the existing
SharedBytes.copyToCacheFileAlignedthat is not limited to copy only a givenlengthof bytes but copies all bytes that can be read from theInputStream.It also adds a variant of the existing
BlobCacheUtils.computeRange()method that computes range of bytes to expand over a complete regions, instead of being limited to the length of the blob.Finally, it removes an assertion that a
CacheFilecannot write a range larger than the blob length.