Large in-memory S3 transfers

[asylvest]

I've got a suite of applications that produce very large binary buffers that I'd like to be able to intermittently write to and read from S3 with. I'd like to avoid large reallocations and copying of data and instead transfer the buffer directly to/from S3.

The PutObjectRequest wants an IOStream and the GetOjectRequest will allow you to provide a factory that creates an IOStream. So the best workaround I've got so far is to create a StringStream and use pubsetbuf() to override the stream's underlying buffer (in the snippets below, I've got void* buffer and size_t bufferSize).

For the put:

    std::shared_ptr<Aws::StringStream> stream =
            Aws::MakeShared<Aws::StringStream>(ALLOCATION_TAG);

    stream->rdbuf()->pubsetbuf(static_cast<char*>(const_cast<void*>(buffer)),
                               bufferSize);
    stream->rdbuf()->pubseekpos(bufferSize);

    stream->seekg(0);

And for the get:

    request.SetResponseStreamFactory(
            [buffer, bufferSize]()
            {
                std::unique_ptr<Aws::StringStream>
                        stream(Aws::New<Aws::StringStream>(ALLOCATION_TAG));

                stream->rdbuf()->pubsetbuf(static_cast<char*>(buffer),
                                           bufferSize);

                return stream.release();
            });

This appears to work with the testing I've done so far, and as far as I understand streams it seems like this should accomplish what I want in terms of avoiding large allocations or copies. Just wondering if you have any thoughts on an easier/better way to do this or have any plans to add any overrides in the API to make this a little easier.

Also, to confirm, a single PutObjectRequest/GetObjectRequest will automatically do multi-part uploads for large transfers right? And if I don't override the rate limiter, by default it'll use as much network bandwidth as it can get?

Thanks.
-Adam

[JonathanHenson]

Have you looked at aws-sdk-cpp-transfer? It does many of the things you are
trying to do. It will allocate some buffers upfront, but afterwards will
just do copies. It will also do multipart for you automatically if the file
is over 5mb in size.

aws-sdk-cpp is a low-level client. Meaning, it is just an interface for
talking to the s3 service and nothing more.

What you have should work at first glance, hut the transfer manager is
designed for just this purpose.

Also, you are correct that if you dont override the rate limiter, then
everything will he wide open.

We discuss the transfer manager in the video in the readme. Its the last 15
minutes or so.
On Jan 3, 2016 7:49 AM, "Adam Sylvester" notifications@github.com wrote:

I've got a suite of applications that produce very large binary buffers
that I'd like to be able to intermittently write to and read from S3 with.
I'd like to avoid large reallocations and copying of data and instead
transfer the buffer directly to/from S3.

The PutObjectRequest wants an IOStream and the GetOjectRequest will allow
you to provide a factory that creates an IOStream. So the best workaround
I've got so far is to create a StringStream and use pubsetbuf() to
override the stream's underlying buffer (in the snippets below, I've got void*
buffer and size_t bufferSize).

For the put:

std::shared_ptr<Aws::StringStream> stream =
        Aws::MakeShared<Aws::StringStream>(ALLOCATION_TAG);

stream->rdbuf()->pubsetbuf(static_cast<char*>(const_cast<void*>(buffer)),
                           bufferSize);
stream->rdbuf()->pubseekpos(bufferSize);

stream->seekg(0);

And for the get:

request.SetResponseStreamFactory(
        [buffer, bufferSize]()
        {
            std::unique_ptr<Aws::StringStream>
                    stream(Aws::New<Aws::StringStream>(ALLOCATION_TAG));

            stream->rdbuf()->pubsetbuf(static_cast<char*>(buffer),
                                       bufferSize);

            return stream.release();
        });

This appears to work with the testing I've done so far, and as far as I
understand streams it seems like this should accomplish what I want in
terms of avoiding large allocations or copies. Just wondering if you have
any thoughts on an easier/better way to do this or have any plans to add
any overrides in the API to make this a little easier.

Also, to confirm, a single PutObjectRequest/GetObjectRequest will
automatically do multi-part uploads for large transfers right? And if I
don't override the rate limiter, by default it'll use as much network
bandwidth as it can get?

Thanks.
-Adam

—
Reply to this email directly or view it on GitHub
#64.

[JonathanHenson]

Correction

aws-sdk-cpp-s3* is a low-level client.
On Jan 3, 2016 8:07 AM, "Jonathan Henson" jonathan.michael.henson@gmail.com
wrote:

Have you looked at aws-sdk-cpp-transfer? It does many of the things you
are trying to do. It will allocate some buffers upfront, but afterwards
will just do copies. It will also do multipart for you automatically if the
file is over 5mb in size.

aws-sdk-cpp is a low-level client. Meaning, it is just an interface for
talking to the s3 service and nothing more.

What you have should work at first glance, hut the transfer manager is
designed for just this purpose.

Also, you are correct that if you dont override the rate limiter, then
everything will he wide open.

We discuss the transfer manager in the video in the readme. Its the last
15 minutes or so.
On Jan 3, 2016 7:49 AM, "Adam Sylvester" notifications@github.com wrote:

I've got a suite of applications that produce very large binary buffers
that I'd like to be able to intermittently write to and read from S3 with.
I'd like to avoid large reallocations and copying of data and instead
transfer the buffer directly to/from S3.

The PutObjectRequest wants an IOStream and the GetOjectRequest will allow
you to provide a factory that creates an IOStream. So the best workaround
I've got so far is to create a StringStream and use pubsetbuf() to
override the stream's underlying buffer (in the snippets below, I've got void*
buffer and size_t bufferSize).

For the put:

std::shared_ptr<Aws::StringStream> stream =
        Aws::MakeShared<Aws::StringStream>(ALLOCATION_TAG);

stream->rdbuf()->pubsetbuf(static_cast<char*>(const_cast<void*>(buffer)),
                           bufferSize);
stream->rdbuf()->pubseekpos(bufferSize);

stream->seekg(0);

And for the get:

request.SetResponseStreamFactory(
        [buffer, bufferSize]()
        {
            std::unique_ptr<Aws::StringStream>
                    stream(Aws::New<Aws::StringStream>(ALLOCATION_TAG));

            stream->rdbuf()->pubsetbuf(static_cast<char*>(buffer),
                                       bufferSize);

            return stream.release();
        });

This appears to work with the testing I've done so far, and as far as I
understand streams it seems like this should accomplish what I want in
terms of avoiding large allocations or copies. Just wondering if you have
any thoughts on an easier/better way to do this or have any plans to add
any overrides in the API to make this a little easier.

Also, to confirm, a single PutObjectRequest/GetObjectRequest will
automatically do multi-part uploads for large transfers right? And if I
don't override the rate limiter, by default it'll use as much network
bandwidth as it can get?

Thanks.
-Adam

—
Reply to this email directly or view it on GitHub
#64.

[asylvest]

Thanks - I hadn't noticed that. Looking at TransferClient, I only see an interface for uploading and downloading files from disk by filename... am I missing an API that allows you to do this with in-memory buffers (i.e. I have a void* of a known length that I want to send directly to S3)? That being said, even if TransferClient doesn't support this, I think it will be a very helpful example to help me fine-tune the code I'm working on.

[JonathanHenson]

have a look here and work backwards, you should be able to reuse our resource management interface.

https://github.com/awslabs/aws-sdk-cpp/blob/master/aws-cpp-sdk-transfer/include/aws/transfer/TransferClientDefs.h

[asylvest]

Thanks for the pointer. Does the C++ SDK support multipart downloads? I see multipart upload requests in sdk-s3 and examples of them being used both in sdk-s3-integration-tests and sdk-transfer but don't see any reference to multipart downloads anywhere (or an option to specify this in the GetObjectRequest.

Am I just missing this? If not, any guesses as to when the SDK will be updated to support it?

[JonathanHenson]

Each service client supports whatever a given service exposes, they are
generated directly from the service models. As to multi-part download,
http://docs.aws.amazon.com/AmazonS3/latest/API/RESTObjectGET.html

I recommend reading the api docs for s3. Once you determine what low-level
operations you want to do, you can then use the client to perform that
operation.
On Jan 10, 2016 8:46 AM, "Adam Sylvester" notifications@github.com wrote:

Thanks for the pointer. Does the C++ SDK support multipart downloads? I
see multipart upload requests in sdk-s3 and examples of them being used
both in sdk-s3-integration-tests and sdk-transfer but don't see any
reference to multipart downloads anywhere (or an option to specify this in
the GetObjectRequest.

Am I just missing this? If not, any guesses as to when the SDK will be
updated to support it?

—
Reply to this email directly or view it on GitHub
#64 (comment).

[asylvest]

Thanks... I should have taken a closer look. I am successfully using the S3Client::SetRange() method and everything is working as expected. In terms of potential future enhancements, it would be a little easier to use the SDK if it supported passing in some type of C++ range object rather than a string, but now that I know how this works it is easy enough for me to wrap this in my own method and build up the string under the hood.

[cugarteblair]

In your example above with the StringStream for the "get" how do you access the underlying stream?
If I understand this pattern correctly:
Using the lambda you set the output to a StringStream using request.SetResponseStreamFactory.
Then call:
auto getObjectOutcome = Client->GetObject(request);
where the call to GetObject blocks. So how do you read the stream after that?
If I use a file (vice stream) I see I can read the file. But I am having trouble figuring out how to do it with a stream.

Any help would be appreciated.

[asylvest]

So the cool thing is that you don't need to 'get' the underlying stream at all. When the AWS SDK uses the StringStream to write, it's writing directly into the void* buffer that you provide to the lambda. So, you've got the contents of the S3 object right in buffer... no need to get at the StringStream object itself.

To clarify a bit though, when I want the entire S3 object in memory, I first ask S3 the size of the object, then allocate a buffer / resize a vector to be that size, then pass that pointer to the lambda. buffer has to be at least as large as the S3 object or something bad's going to happen (I guess I'm not sure offhand what'll happen if you did the pubsetbuf() with bufferSize < s3ObjectSize, but I am guessing either the StringStream write would block or something would index OOB).

I've been using this successfully so far but of course would like to get some more miles on it in case there's a shortcoming I haven't run into yet, so let me know if you have any issues with the approach.

[cugarteblair]

Thanks for the quick reply. Your answer is really helpful.

This seems like it would work when one can read the entire object in memory.
I was hoping to find a way to stream large objects processing them as they are being read without waiting for the entire object to be read into memory. In this way it doesn't matter how big the object is and with a multi-threaded reader one can do parallel reads to several objects at once.
What this will do for me is reduce the latency of the request/response in a streaming server I am writing.
So in short I am trying to figure out how to attach to the underlying stream object so I can de-serialize the google protocol buffer objects being streamed.

[asylvest]

Sure thing.

Yep - something similar to what you're saying is on my todo list too. You can use S3Client::SetRange() to specify just a portion of the object that you want. The C++ SDK is thread-safe. So you could check up front what the object size is and either allocate one big buffer that all threads share or allocate a per-thread buffer at whatever size you want then kick off a pool of threads that read a desired range of the object (via SetRange() and the pubsetbuf() stuff above) and do whatever processing you need. I haven't had a chance to implement this yet myself but I think all the pieces should play nice together.

[cugarteblair]

Thanks for your help. I'll look into it. And report back if I have anything of value to add.

But I think I am solving a slightly different problem:
Is there a way to pass in a stream with dynamically allocated memory that the S3 api can write to while my thread reads from it?

I am thinking my server (acting as an s3 client) would be reading several large objects at once streaming them back while processing them and putting the result of the processing on separate dedicated queues that get drained in proper order by a sender thread. This sender thread would be sending responses back to my client.

So it's not necessarily breaking up a large object, but streaming it. And I don't think it would be a problem if the memory is dynamically allocated, that would actually be easier if the stream would handle it as basic IO streams do.

It seems to me the advantage of streaming (in general) is being able to process the head of an arbitrarily sized (and possibly huge) stream at the client side while the server is still sending the tail portion. Since I am processing packets out of the S3 object and sending a set of response messages for each this would reduce the latency of the request/response between my client and server because I don't have to wait until I receive the entire object before sending response messages.
I guess I'm stuck on the notion of reading the object as a dynamic stream. I don't mind the memory allocations.

[asylvest]

Ahh gotcha... in that case, you should be able to just send the SDK a normal StringStream (without the pubsetbuf()). The SDK would write to the stream and the StringStream would internally dynamically allocate more space as needed. You want to be able to access the stream afterwards, so you create the stream beforehand and make a lambda for SetResponseStreamFactory that just returns a pointer to that stream. You keep the stream in scope and then just access it directly after the SDK call.

[cugarteblair]

That is what I have been trying to do but to no avail.

here is the compiler output:
../src/mnem_extract_server.cpp: In member function ‘void AwsS3DataMoverPub::processPktFile()’: ../src/mnem_extract_server.cpp:197:17: error: no matching function for call to ‘Aws::S3::Model::GetObjectRequest::SetResponseStreamFactory(AwsS3DataMoverPub::processPktFile()::__lambda4)’ });

For the following code:

This is defined in the class protected data:
std::shared_ptr<Aws::StringStream> s3_stream;

This is in the Ctor of the class:

s3_stream = Aws::MakeShared<Aws::StringStream>(ALLOCATION_TAG);

And finally this is the call to the lambda using a local scope variable:

std::shared_ptr<Aws::StringStream> m_stream = s3_stream;
getObjectRequest.SetResponseStreamFactory(
                          [m_stream,obj_name]()
                          {
                        cout << obj_name << " - lambda called with stream" << endl;
                        return m_stream;
                          });

[asylvest]

I just skimmed over this, but take a look at the response stream factory signature... I'm pretty sure it wants a raw rather than shared pointer (so change to return m_stream.get()).

[cugarteblair]

Thanks!
I really appreciate the time you are taking to help me out on this.

I'm wasn't sure if this needed to be a shared pointer or not. I was following the pattern from the other stuff. I'm new to the smart pointer concept.

The following code compiles _but is it ok at runtime?_

I assume I own the stream but how do I know the api thread that is using it is done? What if I want to implement a "cancel" ? It seems that a smart pointer might be needed in this case.
I'm surprised the factory signature uses a raw pointer:

The lamda call:
Aws::StringStream* m_stream = s3_stream; getObjectRequest.SetResponseStreamFactory([m_stream]() { return m_stream; });

In the Ctor:
s3_stream = new Aws::StringStream(ALLOCATION_TAG);

Where s3_stream was defined in protected data as:

Aws::StringStream* s3_stream;

[JonathanHenson]

The memory returned by the reaponsestreamfactory will be deleted when the
result object goes out of scope. Don't pass underlying smart pointer memory
to it.
On Jan 14, 2016 4:39 AM, "Adam Sylvester" notifications@github.com wrote:

I just skimmed over this, but take a look at the response stream factory
signature... I'm pretty sure it wants a raw rather than shared pointer (so
change to return m_stream.get()).

—
Reply to this email directly or view it on GitHub
#64 (comment).

[cugarteblair]

@JonathanHenson
A couple of questions:
From the examples I don't see a result object. Is that visible to the user? i.e. can I control the scope. Or are you saying this is a bad idea?

From the example:

GetObjectRequest getObjectRequest;
getObjectRequest.SetBucket(fullBucketName);
getObjectRequest.SetKey(keyName);
getObjectRequest.SetResponseStreamFactory([](){ 
   return Aws::New<Aws::FStream>(ALLOCATION_TAG,DOWNLOADED_FILENAME, std::ios_base::out ); 
                                 });
auto getObjectOutcome = s3Client->GetObject(getObjectRequest);

If I understand this correctly GetObject blocks until the transfer is complete.
Is there anyway to interface to this API asynchronously so I can initiate a transfer into a stream and in a separate thread read data out of the stream? This way I can process data immediatly

[JonathanHenson]

Result object is inside the outcome. e.g. getObjectOutcome.GetResult();

You can take ownership of the Result via move semantics. In that scenario, the memory is yours once you call GetResultWithOwnership();

All low-level api calls have async and future variants. So you can always use those.

[chebz]

I can't get this to work... I have a char* array with some eof characters (.png file, but I would like not use FStream), I can't figure out how to send it. I can't use StringStream, because of eof. The solution that was originally suggested by OP doesn't work for me, I just get timeout error.

I've also tried creating IOStream like described here: http://stackoverflow.com/questions/7781898/get-an-istream-from-a-char

But that doesn't work either.

[JonathanHenson]

This is currently being reviewed internally, but I'd love your thoughts while we're at it:

https://s3.amazonaws.com/aws-sdk-cpp-demos/transferRewrite.diff

Notes:

Deleted TransferClient and everything related to it. Added a simpler interface, TransferManager. TransferHandle is the other half of the control interface which is what the user would use to control their transfers. I added a few utilities to core to support the changes. Largely a generic queue locker, and a std::stream_buf that runs off of a preallocated buffer.

Later I'll be adding parallel downloads (harder to pull off), and directory synchronization, but this has feature parity with the original TransferClient.

Major performance differences:
All stream memory is preallocated (so no more allocation during the transfer)
Threading model is simplified.
No busy waits

Things I'm not happy about:
Mutex heavy
I discovered that we aren't handling Http/Continue correctly on Windows while testing. however, this isn't a regression from the previous code base so I'll add it to the backlog.
For multi-part uploads, we only update the transfer progress for each part because it was too complicated to back out updates if a part fails.

Testing notes:
all unit tests, and integration tests pass.
Uploads saturate the nic (which was my goal)
Downloads, since they aren't in parallel, still need work from a performance standpoint.

[hailbird]

S3Client has no member named "SetRange", right?

error: 'class Aws::S3::S3Client' has no member named 'SetRange'
  context->client->SetRange(range.str());
                   ^

[JonathanHenson]

See GetObjectRequest::SetRange

…

Sent from my iPhone

On Dec 6, 2016, at 7:52 AM, hailbird ***@***.***> wrote: S3Client has no member named "SetRange", right? error: 'class Aws::S3::S3Client' has no member named 'SetRange' context->client->SetRange(range.str()); ^ — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub, or mute the thread.

[hailbird]

I tried it before I asked the question, but failed. See issue-366 for detail. I checked the signature is correct by enhancing libs3 and do the same thing. The enhaced libs3 with v4 support can get the range of the object.

[RPCPP]

Hi,

In advance sorry if this is not the right thread for my question.

I have a question about TransferManager UploadFile function. I am able to transfer 5Mb file, but when I tried to upload bigger file 7MB or more. I get "[WARN]AWSErrorMarshaller [72860] Encountered Unknown AWSError
XML_ERROR_MISMATCHED_ELEMENT
Unable to generate a proper httpResponse from the response stream. Response code: 503:" error.

What does this error means?

[AcidLeroy]

I am having similar problems with the AwsTransfer library, particularly on the "DownloadFile" function. Ideally, I do not want to write to a file and then have to read the contents back into memory. Your integration tests here: https://github.com/aws/aws-sdk-cpp/blob/master/aws-cpp-sdk-transfer-tests/TransferTests.cpp#L194-L235 check the uploaded file by downloading to an FStream. I am able to successfully do this; however, I have not been able to successfully implement the this to an std::stringstream. For small transfers (less than 20MB) I don't seem to have problems, but when I increase the size to 21MB, I no longer get the entire file. Here is my attempt:

std::shared_ptr<std::stringstream> AwsTransfer::SyncDownload(const std::string& bucket, const std::string& key) {
    std::shared_ptr<std::stringstream> contents = std::make_shared<std::stringstream>();
    std::shared_ptr<TransferHandle> request = m_transfer->DownloadFile(bucket, key, [&contents]() {
        std::iostream* out = Aws::New<std::iostream>(allocationTag, contents->rdbuf());
        return out;
    });

    request->WaitUntilFinished();

    return contents;
}

Do I need to preallocate the size of the stringstream before passing?

[taddygrrr]

The answer to the original question isn't correct or helpful (ie, use transfer manager). In my experience, you cannot use stringstream to upload binary data; at least, not in the way documented above, with pubsetbuf, pubseekpos and seekg(0).

Use boost bufferstream. See #533

[JonathanHenson]

we're about to kill of std::iostream usage in favor of a simpler interface. Iostream will be supported but wrapped. Hopefully this will make the whole issue go away.

[JonathanHenson]

kill off*

[taddygrrr]

So, checking back; is there a simpler interface? Seems like people were still having same problem in January 18. See #785.