mirrors
/
elasticsearch
mirror of https://mirror.osredm.com/root/elasticsearch.git
2
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Document read-after-write semantics for `getRegister` (#131522) 

Clarifies in its documentation that `BlobContainer#getRegister` offers
only read-after-write semantics rather than full linearizability, and
adds comments to its callers justifying why this is still safe.
This commit is contained in:
David Turner 2025-07-21 11:04:31 +01:00 committed by GitHub
parent a692cbd0b0
commit 888e9a26e4
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
5 changed files with 28 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
modules/repository-s3/src/main/java/org/elasticsearch/repositories/s3/S3BlobContainer.java
View File

@ -830,7 +830,13 @@ class S3BlobContainer extends AbstractBlobContainer {
.<Void>newForked(l -> ensureOtherUploadsComplete(uploadId, uploadIndex, currentUploads, l))
// Step 4: Read the current register value.
// Step 4: Read the current register value. Note that getRegister only has read-after-write semantics but that's ok here as:
// - all earlier uploads are now complete,
// - our upload is not completing yet, and
// - later uploads can only be completing if they have already aborted ours.
// Thus if our operation ultimately succeeds then there cannot have been any concurrent writes in flight, so this read
// cannot have observed a stale value, whereas if our operation ultimately fails then it doesn't matter what this read
// observes.
.<OptionalBytesReference>andThen(l -> getRegister(purpose, rawKey, l))

19
server/src/main/java/org/elasticsearch/common/blobstore/BlobContainer.java
View File

@ -304,7 +304,10 @@ public interface BlobContainer {
/**
* Atomically sets the value stored at the given key to {@code updated} if the {@code current value == expected}.
* Keys not yet used start at initial value 0. Returns the current value (before it was updated).
* If a key has not yet been used as a register, its initial value is an empty {@link BytesReference}.
* <p>
* This operation, together with {@link #compareAndSetRegister}, must have linearizable semantics: a collection of such operations must
* act as if they operate serially, with each operation taking place at some instant in between its invocation and its completion.
*
* @param purpose The purpose of the operation
* @param key key of the value to update
@ -323,9 +326,12 @@ public interface BlobContainer {
/**
* Atomically sets the value stored at the given key to {@code updated} if the {@code current value == expected}.
* Keys not yet used start at initial value 0.
* If a key has not yet been used as a register, its initial value is an empty {@link BytesReference}.
* <p>
* This operation, together with {@link #compareAndExchangeRegister}, must have linearizable semantics: a collection of such operations
* must act as if they operate serially, with each operation taking place at some instant in between its invocation and its completion.
*
* @param purpose
* @param purpose The purpose of the operation
* @param key key of the value to update
* @param expected the expected value
* @param updated the new value
@ -350,7 +356,12 @@ public interface BlobContainer {
/**
* Gets the value set by {@link #compareAndSetRegister} or {@link #compareAndExchangeRegister} for a given key.
* If a key has not yet been used, the initial value is an empty {@link BytesReference}.
* If a key has not yet been used as a register, its initial value is an empty {@link BytesReference}.
* <p>
* This operation has read-after-write consistency with respect to writes performed using {@link #compareAndExchangeRegister} and
* {@link #compareAndSetRegister}, but does not guarantee full linearizability. In particular, a {@code getRegister} performed during
* one of these write operations may return either the old or the new value, and a caller may therefore observe the old value
* <i>after</i> observing the new value, as long as both such read operations take place before the write operation completes.
*
* @param purpose The purpose of the operation
* @param key key of the value to get

6
server/src/main/java/org/elasticsearch/common/blobstore/support/BlobContainerUtils.java
View File

@ -33,9 +33,9 @@ public class BlobContainerUtils {
}
/**
* Many blob stores have consistent (linearizable/atomic) read semantics and in these casees it is safe to implement {@link
* BlobContainer#getRegister} by simply reading the blob using this utility.
*
* Many blob stores have consistent read-after-write semantics and in these cases it is safe to implement
* {@link BlobContainer#getRegister} by simply reading the blob using this utility.
* <p>
* NB it is not safe for the supplied stream to resume a partial downloads, because the resumed stream may see a different state from
* the original.
*/

2
x-pack/plugin/snapshot-repo-test-kit/src/main/java/org/elasticsearch/repositories/blobstore/testkit/analyze/ContendedRegisterAnalyzeAction.java
View File

@ -156,6 +156,8 @@ class ContendedRegisterAnalyzeAction extends HandledTransportAction<ContendedReg
};
if (request.getInitialRead() > request.getRequestCount()) {
// This is just the initial read, so we can use getRegister() despite its weaker read-after-write semantics: all subsequent
// operations of this action use compareAndExchangeRegister() and do not rely on this value being accurate.
blobContainer.getRegister(OperationPurpose.REPOSITORY_ANALYSIS, registerName, initialValueListener.delegateFailure((l, r) -> {
if (r.isPresent()) {
l.onResponse(r);

1
x-pack/plugin/snapshot-repo-test-kit/src/main/java/org/elasticsearch/repositories/blobstore/testkit/analyze/RepositoryAnalyzeAction.java
View File

@ -651,6 +651,7 @@ public class RepositoryAnalyzeAction extends HandledTransportAction<RepositoryAn
case 0 -> new CheckedConsumer<ActionListener<OptionalBytesReference>, Exception>() {
@Override
public void accept(ActionListener<OptionalBytesReference> listener) {
// All register operations have completed by this point so getRegister is safe
getBlobContainer().getRegister(OperationPurpose.REPOSITORY_ANALYSIS, registerName, listener);
}