Fix read corruption in S3 (#2253)

This fixes a rare read corruption that may manifest with multiple different error
messages. One example:
`[TileDB::Buffer] Error: Read failed; Trying to read beyond buffer size`

The issue is that the `Aws::IOStream` constructor that we use within the callback
set in `GetObjectRequest::SetResponseStreamFactory()` accepts a pointer to a
`boost::interprocess::bufferbuf`. We currently allocate one of these `bufferbuf`
instances on the heap and free it when we return from `S3::read`. Experimentally,
I have determined that this can cause a corruption to the buffer we stream into.

We are not responsible for what the AWS SDK does with objects created from the
callback we assigned in `GetObjectRequest::SetResponseStreamFactory()`. This
patch introduces a small wrapper so that the `Aws::IOStream` manages the lifetime
of the `boost::interprocess::bufferbuf`.

I also noticed that we use an SDK version with a `Aws::Utils::Stream::PreallocatedStreamBuf`
that we can use instead of the boost `bufferbuf`.

Relevant discussion on stack overflow:
aws/aws-sdk-cpp#1430

---

TYPE: BUG
DESC: Fix rare read corruption in S3

Co-authored-by: Joe Maley <[email protected]>

Author: joe maley

tiledb/sm/filesystem/s3.cc

@@ -762,16 +762,14 @@ Status S3::read(
       ("bytes=" + std::to_string(offset) + "-" +
        std::to_string(offset + length + read_ahead_length - 1))
           .c_str());
-  // Create a unique_ptr so that this will be freed at the end of the function
-  // call This only needs to live long enough for the request itself
-  auto streamBuf = tdb_unique_ptr<boost::interprocess::bufferbuf>(tdb_new(
-      boost::interprocess::bufferbuf,
-      (char*)buffer,
-      length + read_ahead_length));
-  get_object_request.SetResponseStreamFactory([&streamBuf]() {
-    return Aws::New<Aws::IOStream>(
-        constants::s3_allocation_tag.c_str(), streamBuf.get());
-  });
+  get_object_request.SetResponseStreamFactory(
+      [buffer, length, read_ahead_length]() {
+        return Aws::New<PreallocatedIOStream>(
+            constants::s3_allocation_tag.c_str(),
+            buffer,
+            length + read_ahead_length);
+      });
+
   if (request_payer_ != Aws::S3::Model::RequestPayer::NOT_SET)
     get_object_request.SetRequestPayer(request_payer_);
 

tiledb/sm/filesystem/s3.h

@@ -56,6 +56,7 @@
 #include <aws/core/utils/UUID.h>
 #include <aws/core/utils/memory/stl/AWSStringStream.h>
 #include <aws/core/utils/ratelimiter/DefaultRateLimiter.h>
+#include <aws/core/utils/stream/PreallocatedStreamBuf.h>
 #include <aws/core/utils/threading/Executor.h>
 #include <aws/identity-management/auth/STSAssumeRoleCredentialsProvider.h>
 #include <aws/s3/S3Client.h>
@@ -508,6 +509,32 @@ class S3 {
     mutable std::mutex mtx;
   };
 
+  /**
+   * Used to stream results from the GetObject request into
+   * a pre-allocated buffer.
+   */
+  class PreallocatedIOStream : public Aws::IOStream {
+   public:
+    /**
+     * Value constructor. Does not take ownership of the
+     * input buffer.
+     *
+     * @param buffer The pre-allocated underlying buffer.
+     * @param size The maximum size of the underlying buffer.
+     */
+    PreallocatedIOStream(void* const buffer, const size_t size)
+        : Aws::IOStream(new Aws::Utils::Stream::PreallocatedStreamBuf(
+              reinterpret_cast<unsigned char*>(buffer), size)) {
+    }
+
+    /** Destructor. */
+    ~PreallocatedIOStream() {
+      // Delete the unmanaged `Aws::Utils::Stream::PreallocatedStreamBuf`
+      // that was allocated in the constructor.
+      delete rdbuf();
+    }
+  };
+
   /* ********************************* */
   /*         PRIVATE ATTRIBUTES        */
   /* ********************************* */