HADOOP-19354. S3A: S3AInputStream to be created by factory under S3AStore #7214
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
|
test failure from me pushing disk allocator down into store and test case not setting the store up |
steveloughran
force-pushed
the
s3/HADOOP-19354-s3a-inputstream-factory
branch
from
5a32f16 to
7d76047
Compare
steveloughran
force-pushed
the
s3/HADOOP-19354-s3a-inputstream-factory
branch
from
a944b86 to
0f01d61
Compare
|
💔 -1 overall
This message was automatically generated. |
steveloughran
force-pushed
the
s3/HADOOP-19354-s3a-inputstream-factory
branch
from
0f01d61 to
e7e454c
Compare
|
💔 -1 overall
This message was automatically generated. |
| @@ -993,7 +983,7 @@ private void initThreadPools(Configuration conf) { | |||
| unboundedThreadPool.allowCoreThreadTimeOut(true); | |||
| executorCapacity = intOption(conf, | |||
| EXECUTOR_CAPACITY, DEFAULT_EXECUTOR_CAPACITY, 1); | |||
| if (prefetchEnabled) { | |||
| if (requirements.createFuturePool()) { | |||
There was a problem hiding this comment.
change the name to prefetchRequirements.
There was a problem hiding this comment.
there's more requirements than just prefetching, e.g if vector IO support is needed then some extra threads are added to the pool passed down.
...doop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/streams/ObjectInputStreamCallbacks.java
Outdated
Show resolved
Hide resolved
|
I'm just setting this up so it is ready for the analytics stream work...making sure that prefetch is also covered is my way to validate the factory model, and that the options need to include things like the options to ask for a shared thread pool and stream thread pool, with the intent that analytics will use that too. And once I do that, they all need a single base stream class. For my vector IO resilience PR, once I have this PR in, I'm going to go back to #7105 and make it something which works with all object input streams
read failure
the read failure stuff is essentially in my PR, so maybe we can rebase onto this, merge in and then pull up. Goal: analytics stream gets vector IO. |
|
💔 -1 overall
This message was automatically generated. |
rajdchak
reviewed
Jan 15, 2025
| this.ioStatistics = streamStatistics.getIOStatistics(); | ||
| this.inputPolicy = context.getInputPolicy(); | ||
| streamStatistics.inputPolicySet(inputPolicy.ordinal()); | ||
| this.boundedThreadPool = parameters.getBoundedThreadPool(); |
There was a problem hiding this comment.
I see boundedThreadPool is used in S3AInputStream but not in S3APrefetchingInputStream, can we keep boundedThreadPool local to S3AInputStream?
There was a problem hiding this comment.
each stream can declare what it wants thread-pool wise and we will allocate those to them. If they don't want it, they don't get it.
That bounded thread pool passed down is the semaphore pool we also use in uploads. It takes a subset of the shared pool, has its own pending queue and blocks the caller thread when that pending queue is full.
If the analytics stream doesn't currently need it -don't ask for any
But I do want to have the vector IO code to be moved out of S3AInputStream so it can work with the superclass, so all streams get it. These also want a bounded number of threads
rajdchak
reviewed
Jan 15, 2025
|
|
||
| /** | ||
| * A stream of data from an S3 object. | ||
| * The blase class includes common methods, stores |
rajdchak
reviewed
Jan 16, 2025
| * This must be re-invoked after replacing the S3Client during test | ||
| * runs. | ||
| * <p> | ||
| * It requires the S3Store to have been instantiated. | ||
| * @param conf configuration. |
rajdchak
reviewed
Jan 16, 2025
| * @param sharedThreads Number of shared threads to included in the bounded pool. | ||
| * @param streamThreads How many threads per stream, ignoring vector IO requirements. | ||
| * @param createFuturePool Flag to enable creation of a future pool around the bounded thread pool. | ||
| */ |
rajdchak
reviewed
Jan 16, 2025
| @@ -845,7 +826,7 @@ private S3AFileSystemOperations createFileSystemHandler() { | |||
| @VisibleForTesting | |||
| protected S3AStore createS3AStore(final ClientManager clientManager, | |||
| final int rateLimitCapacity) { | |||
| return new S3AStoreBuilder() | |||
| final S3AStore st = new S3AStoreBuilder() | |||
There was a problem hiding this comment.
Nit: rename variable to meaningful name
|
@rajdchak thanks for the comments, will address I do want to pull up the vector IO support, with integration with prefetch and cacheing. For prefetch/caching stream we'd ask for a the requested ranges to be split up into
It'd be good to collect stats on cache hit/miss here, to assess integration of vector reads with ranges. When a list of ranges comes down, there is less need to infer the next range and prefetch, and I'm not actually sure how important cacheing becomes. This is why setting parquet up to use vector IO already appears to give speedups comparable to the analytics stream benchmarks published. what I want is best of both worlds: prefetch of rowgroups from stream inference -and when vector reads come in, statisfy those by returning current/active prefetches, or retrieve new ranges through ranged GET requests. #7105 is where that will go; I've halted that until this is in. And I'll only worry about that integration with prefetched/cached blocks with the analytics stream. |
ahmarsuhail
requested changes
Jan 22, 2025
There was a problem hiding this comment.
Thanks @steveloughran, looks good to me overall. Just need to allow for the ClientManager to be passed into the factory.
| : 0); | ||
| // create an executor which is a subset of the | ||
| // bounded thread pool. | ||
| final SemaphoredDelegatingExecutor pool = new SemaphoredDelegatingExecutor( |
There was a problem hiding this comment.
Just a clarifying question, what is the benefit of creating a new SemaphoredDelegatingExecutor per stream vs just creating this once?
There was a problem hiding this comment.
ok I think I get it, this is basically a way to ensure a single stream instance does not use up too many threads.
...-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/streams/StreamIntegration.java
Show resolved
Hide resolved
...op-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/streams/InputStreamType.java
Show resolved
Hide resolved
|
@ahmarsuhail will look at it. just rebase and review of this; last failure seems VM rather than code |
|
@steveloughran do you want to merge this PR into trunk? Or do you want this to go in via our feature branch?
So either this PR goes into trunk directly, or it can go in as part of the feature branch. |
ahmarsuhail
added a commit
that referenced
this pull request
Jan 27, 2025
Original contributor: @steveloughran in PR #7214
steveloughran
force-pushed
the
s3/HADOOP-19354-s3a-inputstream-factory
branch
from
88ee1d2 to
b5346a1
Compare
|
💔 -1 overall
This message was automatically generated. |
|
Do you think we should fallback if a stream factory fails to load? as if they depend on 3rd party libraries those libs may not be deployed across the cluster Good: something works We can/should add an iostats gauge to indicate which indicates which stream is in use -serve it up in FS and stream |
|
@steveloughran personally think we should throw the failure and not have the fallback. Users of both prefetching input stream and AAL will expect performance benefits from using them, and if the failures are not visible, it'll lead to people thinking those streams aren't any faster. |
|
@ahmarsuhail +1 now, unrelated issue. It looks to me like the jersey update's associated junit stuff has stopped tests being discovered in hadoop-aws. I'm rebasing this PR onto the commit before that one just so I can make progress. Can you check out and build trunk and tell me if your run of the hadoop-aws unit tests run any tests -or is it my setup (across both git clones i have of the repo) |
steveloughran
changed the title
|
@steveloughran I just hit the same issue on my CRT PR, unable to run tests :( |
steveloughran
force-pushed
the
s3/HADOOP-19354-s3a-inputstream-factory
branch
from
b5346a1 to
745492d
Compare
S3 InputStreams are created by a factory class, with the choice of factory dynamically chosen by the option fs.s3a.input.stream.type Supported values: classic, prefetching, analytics. S3AStore * Manages the creation and service lifecycle of the chosen factory, as well as forwarding stream construction requests to the chosen factory. * Provides the callbacks needed by both the factories and input streams. * StreamCapabilities.hasCapability(), which is relayed to the active factory. This avoids the FS having to know what capabilities are available in the stream.
steveloughran
force-pushed
the
s3/HADOOP-19354-s3a-inputstream-factory
branch
from
745492d to
9c8e753
Compare
Ability to create custom streams (type = custom), which
reads class from "fs.s3a.input.stream.custom.factory".
This is mainly for testing, especially CNFE and similar.
Unit test TestStreamFactories for this.
ObjectInputStreams save and export stream type to assist
these tests too, as it enables assertions on the generated
stream type.
Simplified that logic related to the old prefetch enabled flag
If fs.s3a.prefetch.enabled is true, the prefetch stream is returned,
the stream.type option is not used at all. Simpler logic, simpler
docs, fewer support calls.
Parameters supplied to ObjectInputStreamFactory.bind converted
to a parameter object. Allows for more parameters to be added later
if ever required.
ObjectInputStreamFactory returns more requirements to
the store/fs. For this reason
StreamThreadOptions threadRequirements();
is renamed
StreamFactoryRequirements factoryRequirements()
VectorIO context changes
* Returned in factoryRequirements()
* exiting configuration reading code moved into
StreamIntegration.populateVectoredIOContext()
* Streams which don't have custom vector IO, e.g. prefetching
can return a minimum seek range of 0.
This disables range merging on the default PositionedReadable
implementation, so ensures that they will only get asked for
data which will be read...leaving prefetch/cache code
to know exactly what is needed.
Other
* Draft docs.
* Stream capability declares stream type
& is exported through FS too.
(todo: test, document, add to bucket-info)
* ConfigurationHelper.resolveEnum() supercedes
Configuration.getEnum() with
- case independence
- fallback is a supplier<Enum> rather than a simple
value.
Change-Id: I2e59300af48042df8173de61d0b3d6139a0ae7fe
Not fully tested yet. I want to have the stream type passed down as a -D option |
|
💔 -1 overall
This message was automatically generated. |
Change-Id: Ib2053402752b05e8388396909251f6cd59bd9cb7
To avoid confusion when output stream support is pushed down.
-Input stream capabilities exported via {inputStreamHasCapability()};
hasCapability() always returns false.
Change-Id: I5836e0c14a6a781b32888baae178883ba47aab9c
TODO: capability probe tests in fs.
|
💔 -1 overall
This message was automatically generated. |
|
🎊 +1 overall
This message was automatically generated. |
StreamFactoryRequirements takes a list of requirement flags, and moves off boolean flags. (incompatible change). A new flag indicates that the stream may issue requests outside an audit span. This is used when initializing the audit manager so out-of-span requests are never rejected. The requirement on vector IO is removed, instead set the #of required threads in VectoredIOContext to zero (may revisit this) VectoredIOContext moved to streams package and build() operation makes immutable * factory.bind() can throw an IOE Change-Id: If2e121e6d3c0c6d19ac6f4b8452752a8bfafd3d1
ahmarsuhail
reviewed
Feb 5, 2025
There was a problem hiding this comment.
Thanks @steveloughran,
I had a question about serviceStop()/close(), not at which point the stream factories are acutally closed. For AAL this is quite important, as the individual stream does not hold any resources..but the factory does, so we need to call close on the factory to release any resources on FS close.
Was also thinking about the lazy creation discussion..and wondering if we should make the factory init lazy in S3AStore.serviceInit(), and only create it on the first executeOpen()..rather than asking implementations bind to be lazy. Know this isn't trivial, so maybe not worth it.
| @@ -4354,22 +4257,25 @@ public void close() throws IOException { | |||
| protected synchronized void stopAllServices() { | |||
| try { | |||
| trackDuration(getDurationTrackerFactory(), FILESYSTEM_CLOSE.getSymbol(), () -> { | |||
| closeAutocloseables(LOG, store); | |||
| closeAutocloseables(LOG, getStore()); | |||
There was a problem hiding this comment.
@steveloughran can you explain the lifecycle of things with the Service stuff?
We're doing closeAutocloseables(LOG, getStore()); here, but Store doesn't actually implement AutoCloseable. So I'm not sure if we need to something here (call serviceStop()?)
Also need to figure out what changes I need to make to close our s3SeekableInputStreamFactory, prior to this changes, I was passing in our AAL factory, in this closeAutocloseables(), which would call .s3SeekableInputStreamFactory.close().
| @Override | ||
| public boolean hasPathCapability(final Path path, final String capability) { | ||
| switch (toLowerCase(capability)) { | ||
| case StreamCapabilities.IOSTATISTICS: |
There was a problem hiding this comment.
AAL doesn't have IoStats yet, so this should probably not return true without checking
Comment on lines
+76
to
+77
| case StreamCapabilities.IOSTATISTICS: | ||
| case StreamStatisticNames.STREAM_LEAKS: |
There was a problem hiding this comment.
similar to above, think we should just let the stream factories define what capabilities they have rather than doing it on the parent class. As AAL does not have IoStats or stream leaks right now
| * and {@code start()}. | ||
| * @param factoryBindingParameters parameters for the factory binding | ||
| */ | ||
| void bind(FactoryBindingParameters factoryBindingParameters); |
There was a problem hiding this comment.
discussed offline: this should throw IoException as createAsyncClient() throws IoException
|
💔 -1 overall
This message was automatically generated. |
This is really hard to do; I've had to roll back the initial attempt because it mandated a loop in init auditor -> request factory -> store -> stream -> requirements Proposed: allow the request factory to have its handler callback updatable after creation. It is nominally possible to set a build factory through maven -Dstream=prefetch However, this isn't being picked up as can be seen with runs of -Dstream=custom -Dstream=unknown MUST fail. they currently don't Change-Id: I8343c8c9bf0a8cd1b353c7c8f4cecf7f569a4a28
steveloughran
force-pushed
the
s3/HADOOP-19354-s3a-inputstream-factory
branch
from
88d31d4 to
677eb50
Compare
|
💔 -1 overall
This message was automatically generated. |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
HADOOP-19354
How was this patch tested?
S3 london
For code changes:
LICENSE,LICENSE-binary,NOTICE-binaryfiles?TODO
VectoredIOContext
VectoredIOContext.build() to freeze setters, add a copy()
method to copy it, which is then used to create the copy
passed down to streams.
(via a private constructor which returns a mutable version)
Stream capabilities
[ ] doc
[ ] add unit and Itests through FS.
[ ] storediag
[ ] bucket-info
IOStats
[ ] thread stats context to be saved in ObjectInputStream
Testing.
[ ] The huge file tests should be tuned so each of the different ones uses a different stream, always.
[ ] use a -Dstream="factory name" to choose factory, rather than the -Dprefetch
[ ] if not set, whatever is in auth-keys gets picked up.
[ ] ConfigurationHelper.resolveEnum() tests
[ ] VectorIO context unit tests for prefetch type
Docs
[ ] stream leaks
[ ] thread IOStats/context resetting
open issues
ITestS3AOpenCost#prefetching probe