AWS S3: Add getObjectByRanges to S3 API #2982

gael-ft · 2023-05-23T13:05:53Z

Note I created a MergeOrderedN flow graph, because I didn't find a way to do the same using existing ones.
Happy to remove it if there's a way.

johanandren

Note that trying to control if the upstreams pre-fetch like you try to do is tricky since the sources could always contain their own buffers and start right away.

There's also a risk that the request is triggered right away and the response body stream times out with a subscription timeout from being queued based on the stream consumption speed.

A more safe solution, If I understand correctly what we are aiming for here, might be a ranges.mapAsync(parallelism)(range => fetchEntireRangeIntoFutureByteString)

johanandren · 2023-05-24T07:23:10Z

s3/src/main/scala/akka/stream/alpakka/s3/impl/MergeOrderedN.scala

+  *
+  * '''Cancels when''' downstream cancels
+  */
+@InternalApi private[impl] final class MergeOrderedN[T](val inputPorts: Int, val breadth: Int) extends GraphStage[UniformFanInShape[T, T]] {


ConcatPreFetch or something like that would be more correct , merge in Akka streams generally mean emit in any order.

gael-ft · 2023-05-24T09:47:51Z

Hi @johanandren

Thanks for the review !

Note that trying to control if the upstreams pre-fetch like you try to do is tricky since the sources could always contain their own buffers and start right away.

There's also a risk that the request is triggered right away and the response body stream times out with a subscription timeout from being queued based on the stream consumption speed.

I assume this is the pupose of GraphStages.withDetachedInputs(...) right ?

A more safe solution, If I understand correctly what we are aiming for here, might be a ranges.mapAsync(parallelism)(range => fetchEntireRangeIntoFutureByteString)

Anyway I'll try this as it would be much shorter (no need MergeOrderedN) and looks like it does the same in the end

gael-ft · 2023-05-24T10:01:09Z

Well, that being said, don't like much ranges.mapAsync(parallelism)(range => fetchEntireRangeIntoFutureByteString).

If I am correct, it means no ByteString will be pushed downstream until the range is completely fetched, which would work, but waiting for it to complete is (sometimes) useless.

Maybe GraphStages.withDetachedInputs(...) would solve the problem ?

johanandren · 2023-05-24T12:57:45Z

Good point on not getting the bytes of the first chunk in a streaming fashion. Hmmm. Possibly some combination of mapAsync + adding a buffer to the response byte source and then flatMapConcat to concatenate the results into the stream.

GraphStages.withDetachedInputs does not help as far as I can see, it adds a one element buffer on each input so that they will all be eagerly started.

gael-ft · 2023-05-24T17:13:20Z

Just want to be sure to fully understand the issue to fix it correctly (and to know what I should take care of in the future 😉):

Note that trying to control if the upstreams pre-fetch like you try to do is tricky since the sources could always contain their own buffers and start right away.

If they are, elements will come from the source buffer, not directly from the remote (S3 here), ot sure what is the problem here ?

There's also a risk that the request is triggered right away and the response body stream times out with a subscription timeout from being queued based on the stream consumption speed.

In the MergeOrderedN I am not pulling all inputs ports at the same time.
I only pull breadth inputs ports.
As well, I either push them downstream directly if I can (and should) or I put them in a queue and continue to process new elements.
There is no "wait until something happen to continue processing ongoing sources"

Example:

sources: S1, S2, S3
parallelism: 2

(preStart) -> pulling S1 & S2 only
S1 push & out available -> push downstream and pull it again
S2 push -> buffer and pull it again
S1 push & out not available -> buffer and pull it again
S1 complete (S1 has buffered elements, so we do nothing)
S2 push -> buffer and pull it again
out available and pull -> dequeue buffer of S1 -> S1 completed and buffer empty -> S2 become the "current" and we start to pull S3
out available and pull -> dequeue elements from buffer of S2
...

I can't figure how I could trigger subscription timeout here ?
If source has its own buffer then problem is not there
If source has not we won't pull it until its time comes so no requests body pending subscription

With some imagination ... we could compare it with .concat no ?
Second source won't be pulled until the first completed, whetever the current state of the second source is. Am I missing something?

Good point on not getting the bytes of the first chunk in a streaming fashion. Hmmm. Possibly some combination of mapAsync + adding a buffer to the response byte source and then flatMapConcat to concatenate the results into the stream.

Hmmm ok, I'll play a bit and see if I can find something

johanandren · 2023-05-25T07:07:28Z

All the range requests are done immediately on materialization, then your scheme is to only pull from the first N of the requests and backpressure the rest. The are two potential problems with it:

Any buffer introduced explicitly, or implicitly, for example by an async boundary, will detach the response stream and start consuming it right away for all requests.

If there is no boundary and the backpressure works: if you don't subscribe to those within a timeout (default 1 second, response-entity-subscription-timeout) those request streams will fail.

The only way to make sure the requests are only done N at a time would be to not trigger each request at all until it should run and start fetch data.

gael-ft · 2023-05-26T23:48:53Z

All the range requests are done immediately on materialization

Ok thanks clear enough to me now.

johanandren · 2023-05-29T16:42:00Z

s3/src/main/scala/akka/stream/alpakka/s3/impl/S3Stream.scala

+          val endMarker = Source.single(ByteString("$END$"))
+          getObject(s3Location, Some(br), versionId, s3Headers).concat(endMarker).map(_ -> idx)
+        })
+        .statefulMapConcat(RangeMapConcat)


Composing each range-source with a buffer to allow parallel fetching, and then concatenating the resulting streams to get the resulting bytes out in the right order seems like it would achieve the same but much simpler.

Am I missing something clever that this does?

If I understood correctly, you are thinking about conflate or something similar to buffer range sources.
Something like:

getObject(s3Location, Some(br), versionId, s3Headers).conflate(_ ++ _).concat(endMarker).map(_ -> idx)

As flatMapMerge may emit in any order, I still need the range idx to order (possibly buffered) bytes.
So output item of flatMapMerge will look like (ByteString, Long) and can be in any order (regarding the Long).

How can I order them back, without statefulMapConcat ? Range2 could emit before range1 is complete and range2 could be complete before range1.

Note I am not trying to buffer "next" range, if bytes of the "next" range are pushed, I'll push them directly downstream as buffering those bytes is useless (?).

As well, regarding buffers, was not sure if it was useful to "hard pull" upstreams until parallelism * rangeSize is consumed.
Something like:

//... .statefulMapConcat(RangeMapConcat) // Might be useful to consume elements of all flatMapMerge materialized upstreams .batchWeighted(parallelism * rangeSize, _.size, identity)(_ ++ _)

I was thinking something like

Source(byteRanges) .mapAsync(parallelism)(range => getObjectByRanges(...).buffer(size, Backpressure) ).flatMapConcat(identity)

But ofc that may not be good enough with buffer sized in chunks instead of bytes, we don't have a buffer with weighted size calculation though, maybe batchWeighted could do, not sure.

Hmm can't make it work:

Tried with:

Source(byteRanges) .mapAsync(parallelism)(br => Future.successful( getObject(s3Location, Some(br), versionId, s3Headers).batchWeighted(rangeSize, _.size, identity)(_ ++ _) )) .flatMapConcat(identity)

and

Source(byteRanges) .mapAsync(parallelism)(br => Future.successful( Source.fromMaterializer { case (mat, _) => getObject(s3Location, Some(br), versionId, s3Headers) .preMaterialize()(mat) ._2 .batchWeighted(rangeSize, _.size, identity)(_ ++ _) } )) .flatMapConcat(identity)

But in both situations, ranges are fetched one by one and download perf looks like getObject.
Just like if .mapAsync(P)(_ => someSource).flatMapConcat(identity) was not enough to materalize P sources at the same time.
Leaving us with the flatMapMerge ...

Ah, ofc, they aren't materialized so they can start consume bytes until flatMapConcat:ed, didn't think of that. Pre-materialization creates a running source but the downstream is not materialized until it is used, so you would need to put the batching before preMaterialize.

For the record I created an upstream issue with an idea that could make this kind of thing easier: akka/akka#31958 (continue with the current solution here though)

He-Pin · 2023-07-30T16:07:35Z

@gael-ft I have submited a PR for flatmapConcat with inflight materialization, would you like to do some review, thanks.

He-Pin · 2023-08-27T11:57:21Z

s3/src/main/scala/akka/stream/alpakka/s3/impl/S3Stream.scala

+          case None =>
+            val exc = new NoSuchElementException(s"Object does not exist at location [${s3Location.mkString}]")
+            objectMetadataMat.failure(exc)
+            Source.failed(exc)


This will close the downstream ASAP, do you want to defer it?

It happens after getObjectMetadata result has been pulled so I am not sure of what that implies in this context ?

The idea is that no ObjectMetadata means no S3 object, so I think the source should fail as well as the materialized Future.

Add getObjectByRanges to S3 API

7f7d142

probot-autolabeler bot added the p:aws-s3 label May 23, 2023

johanandren reviewed May 24, 2023

View reviewed changes

Change fetching ranges strategy

f8bb980

gael-ft requested a review from johanandren May 29, 2023 08:04

johanandren reviewed May 29, 2023

View reviewed changes

johanandren mentioned this pull request Jun 2, 2023

flatMapConcat with parallelism akka/akka#31958

Open

He-Pin reviewed Aug 27, 2023

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

AWS S3: Add getObjectByRanges to S3 API #2982

AWS S3: Add getObjectByRanges to S3 API #2982

gael-ft commented May 23, 2023

johanandren left a comment

johanandren May 24, 2023

gael-ft commented May 24, 2023

gael-ft commented May 24, 2023

johanandren commented May 24, 2023

gael-ft commented May 24, 2023 •

edited

johanandren commented May 25, 2023

gael-ft commented May 26, 2023

johanandren May 29, 2023

gael-ft May 29, 2023 •

edited

johanandren May 30, 2023

johanandren May 30, 2023

gael-ft Jun 2, 2023 •

edited

johanandren Jun 2, 2023

johanandren Jun 2, 2023

He-Pin commented Jul 30, 2023

He-Pin Aug 27, 2023

gael-ft Aug 29, 2023 •

edited

AWS S3: Add getObjectByRanges to S3 API #2982

Are you sure you want to change the base?

AWS S3: Add getObjectByRanges to S3 API #2982

Conversation

gael-ft commented May 23, 2023

johanandren left a comment

Choose a reason for hiding this comment

johanandren May 24, 2023

Choose a reason for hiding this comment

gael-ft commented May 24, 2023

gael-ft commented May 24, 2023

johanandren commented May 24, 2023

gael-ft commented May 24, 2023 • edited

johanandren commented May 25, 2023

gael-ft commented May 26, 2023

johanandren May 29, 2023

Choose a reason for hiding this comment

gael-ft May 29, 2023 • edited

Choose a reason for hiding this comment

johanandren May 30, 2023

Choose a reason for hiding this comment

johanandren May 30, 2023

Choose a reason for hiding this comment

gael-ft Jun 2, 2023 • edited

Choose a reason for hiding this comment

johanandren Jun 2, 2023

Choose a reason for hiding this comment

johanandren Jun 2, 2023

Choose a reason for hiding this comment

He-Pin commented Jul 30, 2023

He-Pin Aug 27, 2023

Choose a reason for hiding this comment

gael-ft Aug 29, 2023 • edited

Choose a reason for hiding this comment

gael-ft commented May 24, 2023 •

edited

gael-ft May 29, 2023 •

edited

gael-ft Jun 2, 2023 •

edited

gael-ft Aug 29, 2023 •

edited