Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Buffered upload no longer requires length in sync client #22218

Merged
merged 9 commits into from
Jun 28, 2021
Merged

Buffered upload no longer requires length in sync client #22218

Show file tree
Hide file tree
Changes from 6 commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
b864c9b
BlobParallelUploadOptions takes -1 length
jaschrep-msft Jun 10, 2021
a89c0b8
javadoc syntax
jaschrep-msft Jun 10, 2021
8cce9a3
ported converter changes to datalake
jaschrep-msft Jun 11, 2021
9a54687
ported converter changes to file share
jaschrep-msft Jun 11, 2021
7ee1c3e
deprecated old constructors
jaschrep-msft Jun 11, 2021
83442bf
test retry on new inputstream converter
jaschrep-msft Jun 14, 2021
89de479
poke ci
jaschrep-msft Jun 21, 2021
47b02c2
redid lengthless buffered upload API
jaschrep-msft Jun 23, 2021
7be605e
fixed bad arg check
jaschrep-msft Jun 23, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 13 additions & 4 deletions sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/BlobAsyncClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -526,11 +526,20 @@ buffers is not a common scenario for async like it is in sync (and we already bu
(stream, length) -> uploadFullBlob(blockBlobAsyncClient, stream, length, parallelTransferOptions,
headers, metadata, tags, tier, requestConditions, computeMd5);

Flux<ByteBuffer> data = options.getDataFlux() == null ? Utility.convertStreamToByteBuffer(
options.getDataStream(), options.getLength(),
Flux<ByteBuffer> data = options.getDataFlux();
// no specified length: use azure.core's converter
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

if we have markable stream - is SDK smart enough to detect it and don't buffer at all?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Unfortunately not. Also, though, that is only applicable when maxConcurrency = 1, so we aren't missing out on the biggest optimization.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Does this PR somewhat impairs ability to introduce such optimization?
We have customers asking about memory overhead. We should offer a way to do unbuffered upload of markable stream.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't think this PR introduces any new hurdles for such a feature that didn't already exist. I did the work for that optimization in .NET and the integration aspect was fairly simple.

if (data == null && options.getLength() == -1) {
// We can only buffer up to max int due to restrictions in ByteBuffer.
(int) Math.min(Integer.MAX_VALUE, parallelTransferOptions.getBlockSizeLong()), false)
: options.getDataFlux();
int chunkSize = (int) Math.min(Integer.MAX_VALUE, parallelTransferOptions.getBlockSizeLong());
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

can you add a note here that this is fine since buffered upload does not require a replayable flux?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It's not on this exact line but it's in the method on decision-making that this is buffered

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This chunkSize isn't related to block size right? It's just for the purpose of stream->flux conversion.
I wonder if we should be using block size here.
few things to consider (and please correct me if this is stupid but I have a feeling that his might be reason we've seen higher than expected demand for memory in perf tests):

  • if this results in double buffering, i.e. stream->flux needs memory and then buffered upload needs it as well then in worst case "uploader" might be busy working on uploading last chunk (list of bytebuffers emitted from converter) and converter trying to supply next block - both will hold memory. I believe reactor will make sure at least one next element is getting prepared while recent block is being uploaded. @gapra-msft do you think this makes sense?
  • allocating smaller chunks to feed Flux might be a bit friendlier for allocator and GC. It will have some overhead but on the other hand it's easier to fit multiple smaller arrays into heap rather than finding place for Integer.Max_Value array. The later might trigger GC/heap defragmentation efforts. We shouldn't go crazy small though here but I think int.max is too big. A lot of built in components (like bufferedstreams, http stacks) default to 8KB buffer, so that's the lower boundary, not sure how high we should allow it to be - maybe 64kb? or maybe we need another knob.

Copy link
Member Author

@jaschrep-msft jaschrep-msft Jun 22, 2021 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can you file an issue for this and assign it to me, or does it really need to be addressed now?

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, this can be converted to an issue and addressed outside of this PR. Would you mind doing this?

data = FluxUtil.toFluxByteBuffer(options.getDataStream(), chunkSize);
// specified length (legacy requirement): use custom converter. no marking because we buffer anyway.
} else if (data == null) {
// We can only buffer up to max int due to restrictions in ByteBuffer.
int chunkSize = (int) Math.min(Integer.MAX_VALUE, parallelTransferOptions.getBlockSizeLong());
data = Utility.convertStreamToByteBuffer(
options.getDataStream(), options.getLength(), chunkSize, false);
}

return UploadUtils.uploadFullOrChunked(data, ModelHelper.wrapBlobOptions(parallelTransferOptions),
uploadInChunksFunction, uploadFullBlobFunction);
} catch (RuntimeException ex) {
Expand Down
25 changes: 18 additions & 7 deletions ...-storage-blob/src/main/java/com/azure/storage/blob/options/BlobParallelUploadOptions.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,21 +53,32 @@ public BlobParallelUploadOptions(Flux<ByteBuffer> dataFlux) {
/**
* Constructs a new {@code BlobParalleUploadOptions}.
*
* @param dataStream The data to write to the blob. The data must be markable. This is in order to support retries.
* If the data is not markable, consider opening a {@link com.azure.storage.blob.specialized.BlobOutputStream} and
* writing to the returned stream. Alternatively, consider wrapping your data source in a
* {@link java.io.BufferedInputStream} to add mark support.
* @param length The exact length of the data. It is important that this value match precisely the length of the
* data provided in the {@link InputStream}.
* Use {@link #BlobParallelUploadOptions(InputStream)} instead to supply an InputStream without knowing the exact
* length beforehand.
*
* @param dataStream The data to write to the blob.
* @param length The exact length of the data, or -1. It is important that a non-negative value match precisely the
* length of the data provided in the {@link InputStream}.
* @deprecated length is no longer necessary; use {@link #BlobParallelUploadOptions(InputStream)} instead.
*/
@Deprecated
public BlobParallelUploadOptions(InputStream dataStream, long length) {
StorageImplUtils.assertNotNull("dataStream", dataStream);
StorageImplUtils.assertInBounds("length", length, 0, Long.MAX_VALUE);
StorageImplUtils.assertInBounds("length", length, -1, Long.MAX_VALUE);
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

should we add overload instead of using magic number and use nullable Long to represent this?

well, I just noticed overload below. In that case I'd keep original validation condition here and point users towards other ctor. or maybe create (InputStream, Long) ctor and move this logic there.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is the deprecation not enough to point users to the other ctor? I don't like the idea of introducing another ctor with a Long in it because the whole point of this change it's meaningless to provide one. There's nothing extra bought from providing a length, we're just preserving old behavior in case someone relied on us catching a mismatched length or something like that. A Long implies it may still be of use or provide some optimization or something when it really doesn't.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

then we should either have private ctor or don't call one from the other one.
We should also deprecate getLength and replace it with something like Long getSize() ?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I liked size in the context of expanding the ParallelTransferOptions back then because we could also rename the setter. But here we're not adding a new setter to complete the connection. Perhaps Long getOptionalLength()?

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

either way is fine, we'll know at apireview.

this.dataStream = dataStream;
this.length = length;
this.dataFlux = null;
}

/**
* Constructs a new {@code BlobParalleUploadOptions}.
*
* @param dataStream The data to write to the blob.
*/
public BlobParallelUploadOptions(InputStream dataStream) {
this(dataStream, -1);
}

/**
* Constructs a new {@code BlobParallelUploadOptions}.
*
Expand Down
48 changes: 48 additions & 0 deletions sdk/storage/azure-storage-blob/src/test/java/com/azure/storage/blob/BlobAPITest.groovy
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -209,6 +209,54 @@ class BlobAPITest extends APISpec {
.getValue().getETag() != null
}

def "Upload InputStream no length"() {
when:
bc.uploadWithResponse(new BlobParallelUploadOptions(data.defaultInputStream), null, null)

then:
notThrown(Exception)
bc.downloadContent().toBytes() == data.defaultBytes
}

def "Upload InputStream explicit -1 length"() {
when:
bc.uploadWithResponse(new BlobParallelUploadOptions(data.defaultInputStream, -1), null, null)

then:
notThrown(Exception)
bc.downloadContent().toBytes() == data.defaultBytes
}

def "Upload InputStream bad length"() {
when:
bc.uploadWithResponse(new BlobParallelUploadOptions(data.defaultInputStream, length), null, null)

then:
thrown(Exception)

where:
_ | length
_ | 0
_ | -100
_ | data.defaultDataSize - 1
_ | data.defaultDataSize + 1
}

def "Upload successful retry"() {
given:
def clientWithFailure = getBlobClient(
env.primaryAccount.credential,
bc.getBlobUrl(),
new TransientFailureInjectingHttpPipelinePolicy())

when:
clientWithFailure.uploadWithResponse(new BlobParallelUploadOptions(data.defaultInputStream), null, null)

then:
notThrown(Exception)
bc.downloadContent().toBytes() == data.defaultBytes
}

@LiveOnly
// Reading from recordings will not allow for the timing of the test to work correctly.
def "Upload timeout"() {
Expand Down
50 changes: 50 additions & 0 deletions ...age-blob/src/test/resources/session-records/BlobAPITestUploadInputStreamBadLength[0].json
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,50 @@
{
"networkCallRecords" : [ {
"Method" : "PUT",
"Uri" : "https://REDACTED.blob.core.windows.net/7f5018f307f5018f3aec15131ec2a5fb9dde640e685e?restype=container",
"Headers" : {
"x-ms-version" : "2020-08-04",
"User-Agent" : "azsdk-java-azure-storage-blob/12.13.0-beta.1 (11.0.9; Windows 10; 10.0)",
"x-ms-client-request-id" : "a3d9e138-90e1-4885-8cd7-544bce3e1fca"
},
"Response" : {
"Transfer-Encoding" : "chunked",
"x-ms-version" : "2020-08-04",
"Server" : "Windows-Azure-Blob/1.0 Microsoft-HTTPAPI/2.0",
"eTag" : "0x8D92C58EF0BCAB0",
"Last-Modified" : "Thu, 10 Jun 2021 21:44:32 GMT",
"retry-after" : "0",
"StatusCode" : "201",
"x-ms-request-id" : "69ef6576-b01e-0019-0641-5e7a16000000",
"x-ms-client-request-id" : "a3d9e138-90e1-4885-8cd7-544bce3e1fca",
"Date" : "Thu, 10 Jun 2021 21:44:32 GMT"
},
"Exception" : null
}, {
"Method" : "PUT",
"Uri" : "https://REDACTED.blob.core.windows.net/7f5018f307f5018f3aec15131ec2a5fb9dde640e685e/7f5018f317f5018f3aec68648158188ad9aa3434bb9a",
"Headers" : {
"x-ms-version" : "2020-08-04",
"User-Agent" : "azsdk-java-azure-storage-blob/12.13.0-beta.1 (11.0.9; Windows 10; 10.0)",
"x-ms-client-request-id" : "9fe09bbf-3b98-4f30-b335-6ae82cb45011",
"Content-Type" : "application/octet-stream"
},
"Response" : {
"Transfer-Encoding" : "chunked",
"x-ms-version" : "2020-08-04",
"Server" : "Windows-Azure-Blob/1.0 Microsoft-HTTPAPI/2.0",
"x-ms-content-crc64" : "6RYQPwaVsyQ=",
"Last-Modified" : "Thu, 10 Jun 2021 21:44:33 GMT",
"retry-after" : "0",
"StatusCode" : "201",
"x-ms-request-server-encrypted" : "true",
"Date" : "Thu, 10 Jun 2021 21:44:32 GMT",
"Content-MD5" : "wh+Wm18D0z1D4E+PE252gg==",
"eTag" : "0x8D92C58EF38659F",
"x-ms-request-id" : "69ef6580-b01e-0019-0b41-5e7a16000000",
"x-ms-client-request-id" : "9fe09bbf-3b98-4f30-b335-6ae82cb45011"
},
"Exception" : null
} ],
"variables" : [ "7f5018f307f5018f3aec15131ec2a5fb9dde640e685e", "7f5018f317f5018f3aec68648158188ad9aa3434bb9a" ]
}
50 changes: 50 additions & 0 deletions ...age-blob/src/test/resources/session-records/BlobAPITestUploadInputStreamBadLength[1].json
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,50 @@
{
"networkCallRecords" : [ {
"Method" : "PUT",
"Uri" : "https://REDACTED.blob.core.windows.net/664b29b20664b29b2b8a39735445a8518b952493b840?restype=container",
"Headers" : {
"x-ms-version" : "2020-08-04",
"User-Agent" : "azsdk-java-azure-storage-blob/12.13.0-beta.1 (11.0.9; Windows 10; 10.0)",
"x-ms-client-request-id" : "0a78fd58-091c-48dd-8e06-460e6d3b7b2d"
},
"Response" : {
"Transfer-Encoding" : "chunked",
"x-ms-version" : "2020-08-04",
"Server" : "Windows-Azure-Blob/1.0 Microsoft-HTTPAPI/2.0",
"eTag" : "0x8D92C58EF7CCB72",
"Last-Modified" : "Thu, 10 Jun 2021 21:44:33 GMT",
"retry-after" : "0",
"StatusCode" : "201",
"x-ms-request-id" : "69ef6587-b01e-0019-1241-5e7a16000000",
"x-ms-client-request-id" : "0a78fd58-091c-48dd-8e06-460e6d3b7b2d",
"Date" : "Thu, 10 Jun 2021 21:44:32 GMT"
},
"Exception" : null
}, {
"Method" : "PUT",
"Uri" : "https://REDACTED.blob.core.windows.net/664b29b20664b29b2b8a39735445a8518b952493b840/664b29b21664b29b2b8a255541c44a05dfa854d5d864",
"Headers" : {
"x-ms-version" : "2020-08-04",
"User-Agent" : "azsdk-java-azure-storage-blob/12.13.0-beta.1 (11.0.9; Windows 10; 10.0)",
"x-ms-client-request-id" : "8f4270ac-e7b8-4aba-8fc1-067461d7a2f2",
"Content-Type" : "application/octet-stream"
},
"Response" : {
"Transfer-Encoding" : "chunked",
"x-ms-version" : "2020-08-04",
"Server" : "Windows-Azure-Blob/1.0 Microsoft-HTTPAPI/2.0",
"x-ms-content-crc64" : "6RYQPwaVsyQ=",
"Last-Modified" : "Thu, 10 Jun 2021 21:44:33 GMT",
"retry-after" : "0",
"StatusCode" : "201",
"x-ms-request-server-encrypted" : "true",
"Date" : "Thu, 10 Jun 2021 21:44:33 GMT",
"Content-MD5" : "wh+Wm18D0z1D4E+PE252gg==",
"eTag" : "0x8D92C58EF8BA9BB",
"x-ms-request-id" : "69ef658a-b01e-0019-1441-5e7a16000000",
"x-ms-client-request-id" : "8f4270ac-e7b8-4aba-8fc1-067461d7a2f2"
},
"Exception" : null
} ],
"variables" : [ "664b29b20664b29b2b8a39735445a8518b952493b840", "664b29b21664b29b2b8a255541c44a05dfa854d5d864" ]
}
50 changes: 50 additions & 0 deletions ...age-blob/src/test/resources/session-records/BlobAPITestUploadInputStreamBadLength[2].json
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,50 @@
{
"networkCallRecords" : [ {
"Method" : "PUT",
"Uri" : "https://REDACTED.blob.core.windows.net/4d667a7104d667a71bea6591957c250d6618b480daf9?restype=container",
"Headers" : {
"x-ms-version" : "2020-08-04",
"User-Agent" : "azsdk-java-azure-storage-blob/12.13.0-beta.1 (11.0.9; Windows 10; 10.0)",
"x-ms-client-request-id" : "c79f5d3b-8274-4c59-ada0-c8d13da599cf"
},
"Response" : {
"Transfer-Encoding" : "chunked",
"x-ms-version" : "2020-08-04",
"Server" : "Windows-Azure-Blob/1.0 Microsoft-HTTPAPI/2.0",
"eTag" : "0x8D92C58EFB6BEAF",
"Last-Modified" : "Thu, 10 Jun 2021 21:44:34 GMT",
"retry-after" : "0",
"StatusCode" : "201",
"x-ms-request-id" : "69ef6590-b01e-0019-1a41-5e7a16000000",
"x-ms-client-request-id" : "c79f5d3b-8274-4c59-ada0-c8d13da599cf",
"Date" : "Thu, 10 Jun 2021 21:44:33 GMT"
},
"Exception" : null
}, {
"Method" : "PUT",
"Uri" : "https://REDACTED.blob.core.windows.net/4d667a7104d667a71bea6591957c250d6618b480daf9/4d667a7114d667a71bea705084c8cea16df3a446680d",
"Headers" : {
"x-ms-version" : "2020-08-04",
"User-Agent" : "azsdk-java-azure-storage-blob/12.13.0-beta.1 (11.0.9; Windows 10; 10.0)",
"x-ms-client-request-id" : "4054dcf0-fb43-40f1-8f39-d47e1a4e5500",
"Content-Type" : "application/octet-stream"
},
"Response" : {
"Transfer-Encoding" : "chunked",
"x-ms-version" : "2020-08-04",
"Server" : "Windows-Azure-Blob/1.0 Microsoft-HTTPAPI/2.0",
"x-ms-content-crc64" : "6RYQPwaVsyQ=",
"Last-Modified" : "Thu, 10 Jun 2021 21:44:34 GMT",
"retry-after" : "0",
"StatusCode" : "201",
"x-ms-request-server-encrypted" : "true",
"Date" : "Thu, 10 Jun 2021 21:44:33 GMT",
"Content-MD5" : "wh+Wm18D0z1D4E+PE252gg==",
"eTag" : "0x8D92C58EFC500F4",
"x-ms-request-id" : "69ef6593-b01e-0019-1c41-5e7a16000000",
"x-ms-client-request-id" : "4054dcf0-fb43-40f1-8f39-d47e1a4e5500"
},
"Exception" : null
} ],
"variables" : [ "4d667a7104d667a71bea6591957c250d6618b480daf9", "4d667a7114d667a71bea705084c8cea16df3a446680d" ]
}
50 changes: 50 additions & 0 deletions ...age-blob/src/test/resources/session-records/BlobAPITestUploadInputStreamBadLength[3].json
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,50 @@
{
"networkCallRecords" : [ {
"Method" : "PUT",
"Uri" : "https://REDACTED.blob.core.windows.net/547d4b300547d4b30f11300636363b37d2e8646149bf?restype=container",
"Headers" : {
"x-ms-version" : "2020-08-04",
"User-Agent" : "azsdk-java-azure-storage-blob/12.13.0-beta.1 (11.0.9; Windows 10; 10.0)",
"x-ms-client-request-id" : "d7c66faa-0162-4603-98c4-6fa0af046d3d"
},
"Response" : {
"Transfer-Encoding" : "chunked",
"x-ms-version" : "2020-08-04",
"Server" : "Windows-Azure-Blob/1.0 Microsoft-HTTPAPI/2.0",
"eTag" : "0x8D92C58EFEF5292",
"Last-Modified" : "Thu, 10 Jun 2021 21:44:34 GMT",
"retry-after" : "0",
"StatusCode" : "201",
"x-ms-request-id" : "69ef659a-b01e-0019-2241-5e7a16000000",
"x-ms-client-request-id" : "d7c66faa-0162-4603-98c4-6fa0af046d3d",
"Date" : "Thu, 10 Jun 2021 21:44:33 GMT"
},
"Exception" : null
}, {
"Method" : "PUT",
"Uri" : "https://REDACTED.blob.core.windows.net/547d4b300547d4b30f11300636363b37d2e8646149bf/547d4b301547d4b30f118395439a3b957136742218ef",
"Headers" : {
"x-ms-version" : "2020-08-04",
"User-Agent" : "azsdk-java-azure-storage-blob/12.13.0-beta.1 (11.0.9; Windows 10; 10.0)",
"x-ms-client-request-id" : "96a79c9b-ca06-4499-9825-9a7d0ba61b79",
"Content-Type" : "application/octet-stream"
},
"Response" : {
"Transfer-Encoding" : "chunked",
"x-ms-version" : "2020-08-04",
"Server" : "Windows-Azure-Blob/1.0 Microsoft-HTTPAPI/2.0",
"x-ms-content-crc64" : "6RYQPwaVsyQ=",
"Last-Modified" : "Thu, 10 Jun 2021 21:44:34 GMT",
"retry-after" : "0",
"StatusCode" : "201",
"x-ms-request-server-encrypted" : "true",
"Date" : "Thu, 10 Jun 2021 21:44:33 GMT",
"Content-MD5" : "wh+Wm18D0z1D4E+PE252gg==",
"eTag" : "0x8D92C58EFFD6DF1",
"x-ms-request-id" : "69ef659c-b01e-0019-2341-5e7a16000000",
"x-ms-client-request-id" : "96a79c9b-ca06-4499-9825-9a7d0ba61b79"
},
"Exception" : null
} ],
"variables" : [ "547d4b300547d4b30f11300636363b37d2e8646149bf", "547d4b301547d4b30f118395439a3b957136742218ef" ]
}
Loading