You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "stankiewicz (via GitHub)" <gi...@apache.org> on 2026/07/23 13:41:10 UTC
[PR] Fix ThreadSanitizer data races between bundle processing and async background writer closing in FileBasedSink and WriteFiles [beam]
stankiewicz opened a new pull request, #39458:
URL: https://github.com/apache/beam/pull/39458
Fixes ThreadSanitizer data race errors reported during asynchronous writer closing in `WriteFiles` and `FileBasedSink`.
**Unsynchronized Cross-Thread Lifecycle Access:**
In `WriteFiles.WriteShardsIntoTempFilesFn`, a `FileBasedSink.Writer` is instantiated, opened (`writer.open()`), and written to on the bundle processing thread. Once bundle processing completes, `closeWriterInBackground(writer)` submits asynchronous background tasks (`MoreFutures.runAsync`) to close each writer on a `ForkJoinPool` thread. Because plain Java fields (`id`, `destination`, `outputFile`, `channel`) lack release-acquire barriers across threads, ThreadSanitizer flags reads inside `writer.close()` as racing with writes inside `writer.open()`.
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [ ] Mention the appropriate issue in your description (for example: `addresses #123`), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment `fixes #<ISSUE NUMBER>` instead.
- [ ] Update `CHANGES.md` with noteworthy changes.
- [ ] If this contribution is large, please file an Apache [Individual Contributor License Agreement](https://www.apache.org/licenses/icla.pdf).
See the [Contributor Guide](https://beam.apache.org/contribute) for more tips on [how to make review process smoother](https://github.com/apache/beam/blob/master/CONTRIBUTING.md#make-the-reviewers-job-easier).
To check the build health, please visit [https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md](https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md)
GitHub Actions Tests Status (on master branch)
------------------------------------------------------------------------------------------------
[](https://github.com/apache/beam/actions?query=workflow%3A%22Build+python+source+distribution+and+wheels%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Python+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Java+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Go+tests%22+branch%3Amaster+event%3Aschedule)
See [CI.md](https://github.com/apache/beam/blob/master/CI.md) for more information about GitHub Actions CI or the [workflows README](https://github.com/apache/beam/blob/master/.github/workflows/README.md) to see a list of phrases to trigger workflows.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix ThreadSanitizer data races between bundle processing and async background writer closing in FileBasedSink and WriteFiles [beam]
Posted by "stankiewicz (via GitHub)" <gi...@apache.org>.
stankiewicz commented on PR #39458:
URL: https://github.com/apache/beam/pull/39458#issuecomment-5107512650
@sjvanrossum It's not only open() and async close(), it is also racing between write() and async close. take a look at this approach.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix ThreadSanitizer data races between bundle processing and async background writer closing in FileBasedSink and WriteFiles [beam]
Posted by "sjvanrossum (via GitHub)" <gi...@apache.org>.
sjvanrossum commented on PR #39458:
URL: https://github.com/apache/beam/pull/39458#issuecomment-5127473277
Actually, it looks like `MoreFutures.runAsync(ThrowingRunnable)` submits onto the common `ForkJoinPool` and apparently it does not establish a happens-before relationship between the submitting thread and the executing thread.
What _is_ described is that calling `join()` on a `ForkJoinTask` establishes a happens-before relationship between the executing thread and the joining thread.
Intuitively, I would assume that the same would hold true for submitting/forking threads. I can see why that might not be the case in some circumstances, but I haven't found any reason for it in the documentation (yet).
The `ScheduledExecutorService` provided by the harness does establish a happens-before relationship between the submitting and executing threads. My guess is that this isn't what `ForkJoinPool` is meant for, so I've replaced it with the harness `ScheduledExecutorService` in #39556 which also resolved the TSAN bugs you observed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix ThreadSanitizer data races between bundle processing and async background writer closing in FileBasedSink and WriteFiles [beam]
Posted by "sjvanrossum (via GitHub)" <gi...@apache.org>.
sjvanrossum commented on PR #39458:
URL: https://github.com/apache/beam/pull/39458#issuecomment-5116218821
Ah, of course. We may expect any method to be called on the writer thread and `close`, `writeFooter`, `finishWrite` and `cleanup` on a closer thread.
Any sane implementation of `Channel` should return a value from `isOpen` that's consistent across threads since `close` is documented to guarantee the following behavior ([docs](https://docs.oracle.com/javase/8/docs/api/java/nio/channels/Channel.html#close--)):
>This method may be invoked at any time. If some other thread has already invoked it, however, then another invocation will block until the first invocation is complete, after which it will return without effect.
If you mark `channel` as `volatile` and immediately assign that field to a local variable in `close` and `cleanup`, condition the rest of `close` and `cleanup` logic on `isOpen`, then you're set I think.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix ThreadSanitizer data races between bundle processing and async background writer closing in FileBasedSink and WriteFiles [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #39458:
URL: https://github.com/apache/beam/pull/39458#issuecomment-5060645894
Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment `assign set of reviewers`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix ThreadSanitizer data races between bundle processing and async background writer closing in FileBasedSink and WriteFiles [beam]
Posted by "sjvanrossum (via GitHub)" <gi...@apache.org>.
sjvanrossum commented on PR #39458:
URL: https://github.com/apache/beam/pull/39458#issuecomment-5130265880
Looks like a false positive, see https://github.com/apache/beam/pull/39556#issuecomment-5130227034 for a test case which runs fine without TSAN, but observes a stale read with TSAN.
I suspect that the TSAN integration does not reason through all types of memory modifications and their side effects as specified by the Java Memory Model.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix ThreadSanitizer data races between bundle processing and async background writer closing in FileBasedSink and WriteFiles [beam]
Posted by "stankiewicz (via GitHub)" <gi...@apache.org>.
stankiewicz closed pull request #39458: Fix ThreadSanitizer data races between bundle processing and async background writer closing in FileBasedSink and WriteFiles
URL: https://github.com/apache/beam/pull/39458
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix ThreadSanitizer data races between bundle processing and async background writer closing in FileBasedSink and WriteFiles [beam]
Posted by "gemini-code-assist[bot] (via GitHub)" <gi...@apache.org>.
gemini-code-assist[bot] commented on PR #39458:
URL: https://github.com/apache/beam/pull/39458#issuecomment-5059104945
> [!CAUTION]
> The consumer version of Gemini Code Assist on GitHub has been sunset. All code review activity has officially ceased.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix ThreadSanitizer data races between bundle processing and async background writer closing in FileBasedSink and WriteFiles [beam]
Posted by "sjvanrossum (via GitHub)" <gi...@apache.org>.
sjvanrossum commented on PR #39458:
URL: https://github.com/apache/beam/pull/39458#issuecomment-5102091575
If I understand correctly, `open(String)` initializes all relevant fields which are thereafter never written to again, correct?
Calling `open(String)` concurrently looks unsafe, so ideally it should be marked as `synchronized` along with `close()` to ensure mutually exclusive initialization after construction and to ensure visibility of those writes.
Alternatively, attempt a CAS on `id` before writing any of the other fields for exclusive initialization after construction (note that visibility of the other fields is not yet guaranteed) followed by CAS on either `resourceId` or `channel` to establish a release barrier. That same field must be acquired (e.g., volatile get or `getAcquire`) before reading any other field in `close()`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix ThreadSanitizer data races between bundle processing and async background writer closing in FileBasedSink and WriteFiles [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #39458:
URL: https://github.com/apache/beam/pull/39458#issuecomment-5091718561
Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control. If you'd like to restart, comment `assign set of reviewers`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix ThreadSanitizer data races between bundle processing and async background writer closing in FileBasedSink and WriteFiles [beam]
Posted by "sjvanrossum (via GitHub)" <gi...@apache.org>.
sjvanrossum commented on PR #39458:
URL: https://github.com/apache/beam/pull/39458#issuecomment-5118934942
Something like this perhaps?
I'll run TSAN checks in a bit.
```
/** The channel to write to. */
private volatile @Nullable WritableByteChannel channel;
/**
* Opens a uniquely named temporary file and initializes the writer using {@link #prepareWrite}.
*
* <p>The unique id that is given to open should be used to ensure that the writer's output does
* not interfere with the output of other Writers, as a bundle may be executed many times for
* fault tolerance.
*/
public final void open(String uId) throws Exception {
// Shadow members as local variables to ensure local consistency and force member access
// through this.
final @Nullable String id;
final @Nullable ResourceId outputFile;
final @Nullable WritableByteChannel channel;
try {
this.id = id = spreadUid(uId); // normal store
ResourceId tempDirectory = getWriteOperation().getTempDirectory();
this.outputFile =
outputFile =
tempDirectory.resolve(id, StandardResolveOptions.RESOLVE_FILE); // normal store
verifyNotNull(
outputFile,
"FileSystems are not allowed to return null from resolve: %s",
tempDirectory);
final WritableByteChannelFactory factory =
getWriteOperation().getSink().writableByteChannelFactory;
// The factory may force a MIME type or it may return null, indicating to use the sink's
// MIME.
String channelMimeType = firstNonNull(factory.getMimeType(), mimeType);
CreateOptions createOptions =
StandardCreateOptions.builder()
.setMimeType(channelMimeType)
// The file is based upon a uuid and thus we expect it to be unique and to not
// already
// exist. A new uuid is generated on each bundle processing and thus this also holds
// across bundle retries. Collisions of filenames would result in data loss as we
// would otherwise overwrite already finalized data.
.setExpectFileToNotExist(true)
.build();
WritableByteChannel tempChannel = FileSystems.create(outputFile, createOptions);
try {
this.channel =
channel =
factory.create(tempChannel); // volatile store (sequentially consistent release)
} catch (Exception e) {
// If we have opened the underlying channel but fail to open the compression channel,
// we should still close the underlying channel.
try (tempChannel) {
throw e;
}
}
} catch (Throwable t) {
// Ensure that memory visibility matches non-exception case before throwing.
this.channel = null; // volatile store (sequentially consistent release)
throw t;
}
// The caller shouldn't have to close() this Writer if it fails to open(), so close
// the channel if prepareWrite() or writeHeader() fails.
try {
LOG.debug("Preparing write to {}.", outputFile);
prepareWrite(channel);
LOG.debug("Writing header to {}.", outputFile);
writeHeader();
} catch (Exception e) {
LOG.error("Beginning write to {} failed, closing channel.", outputFile, e);
try (channel) {
throw e;
}
}
LOG.debug("Starting write of bundle {} to {}.", id, outputFile);
}
public final void cleanup() throws Exception {
final @Nullable WritableByteChannel channel = this.channel; // volatile load (acquire)
final @Nullable ResourceId outputFile = this.outputFile; // normal load
try (channel) {
if (channel != null && channel.isOpen()) {
LOG.warn("Channel to {} is still open after cleanup() was called.", outputFile);
}
// channel and outputFile may be null if open() was not called or failed.
if (outputFile != null) {
LOG.info("Deleting temporary file {}", outputFile);
FileSystems.delete(
Collections.singletonList(outputFile), StandardMoveOptions.IGNORE_MISSING_FILES);
}
}
}
/** Closes the channel and returns the bundle result. */
public final void close() throws Exception {
final @Nullable WritableByteChannel channel = this.channel; // volatile load (acquire)
final @Nullable ResourceId outputFile = this.outputFile; // normal load
try (channel) {
checkState(outputFile != null, "FileResult.close cannot be called with a null outputFile");
LOG.debug("Closing {}", outputFile);
writeFooter();
finishWrite();
LOG.debug("Closing channel to {}.", outputFile);
}
LOG.info("Successfully wrote temporary file {}", outputFile);
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix ThreadSanitizer data races between bundle processing and async background writer closing in FileBasedSink and WriteFiles [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #39458:
URL: https://github.com/apache/beam/pull/39458#issuecomment-5089484800
Assigning reviewers:
R: @chamikaramj for label java.
Note: If you would like to opt out of this review, comment `assign to next reviewer`.
Available commands:
- `stop reviewer notifications` - opt out of the automated review tooling
- `remind me after tests pass` - tag the comment author after tests pass
- `waiting on author` - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)
The PR bot will only process comments in the main thread (not review comments).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix ThreadSanitizer data races between bundle processing and async background writer closing in FileBasedSink and WriteFiles [beam]
Posted by "stankiewicz (via GitHub)" <gi...@apache.org>.
stankiewicz commented on PR #39458:
URL: https://github.com/apache/beam/pull/39458#issuecomment-5091701924
R: @sjvanrossum
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix ThreadSanitizer data races between bundle processing and async background writer closing in FileBasedSink and WriteFiles [beam]
Posted by "stankiewicz (via GitHub)" <gi...@apache.org>.
stankiewicz commented on PR #39458:
URL: https://github.com/apache/beam/pull/39458#issuecomment-5114677840
R: @sjvanrossum
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org