github/azure-sdk-for-cpp
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

Make FileBodyStream easy to construct by accepting a file name, and create an internal ParallelFileBodyStream. (#1830)

Browse Source 
Alternative to https://github.com/Azure/azure-sdk-for-cpp/pull/1730 which avoids accepting and breaking the `FILE*` abstraction.
This commit is contained in:
Ahson Khan 2021-03-11 00:39:52 -08:00 committed by GitHub
parent c0ce9688d6
commit 84df3cefbc
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
9 changed files with 325 additions and 143 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
sdk/core/azure-core/CHANGELOG.md
View File

@ -36,6 +36,7 @@
- Introduced `Azure::Core::Http::Policies::LogOptions`, a mandatory parameter for `LogPolicy` construction. Entities that are not specified in the allow lists are hidden in the log.
- Moved `Azure::Core::Logging` namespace entities to `Azure::Core::Diagnostics::Logger` class.
- Removed `Azure::Core::DateTime::GetRfc3339String()`: `Azure::Core::DateTime::ToString()` was extended to provide the same functionality.
- Changed the constructor of `Azure::IO::FileBodyStream` to accept a file name directly and take ownership of opening/closing the file, instead of accepting a file descriptor, offset, and length.
- Renamed the `Range` type to `HttpRange` within the `Azure::Core::Http` namespace.
- Moved `Azure::Core::Response<T>` to `Azure::Response<T>`.
- Moved `Azure::Core::ETag` to `Azure::ETag`.

1
sdk/core/azure-core/CMakeLists.txt
View File

@ -97,6 +97,7 @@ set(
src/http/transport_policy.cpp
src/http/url.cpp
src/io/body_stream.cpp
src/io/random_access_file_body_stream.cpp
src/base64.cpp
src/context.cpp
src/datetime.cpp

130
sdk/core/azure-core/inc/azure/core/io/body_stream.hpp
View File

@ -160,55 +160,125 @@ namespace Azure { namespace Core { namespace IO {
void Rewind() override { m_offset = 0; }
};
namespace _internal {
/**
* @brief A concrete implementation of #Azure::Core::IO::BodyStream used for reading data from
* a file from any offset and length within it.
*/
class RandomAccessFileBodyStream : public BodyStream {
private:
// immutable
#if defined(AZ_PLATFORM_POSIX)
int m_fileDescriptor;
#elif defined(AZ_PLATFORM_WINDOWS)
HANDLE m_filehandle;
#endif
int64_t m_baseOffset;
int64_t m_length;
// mutable
int64_t m_offset;
int64_t OnRead(uint8_t* buffer, int64_t count, Azure::Core::Context const& context) override;
public:
#if defined(AZ_PLATFORM_POSIX)
/**
* @brief Construct from a file descriptor.
*
* @param fileDescriptor A file descriptor to an already opened file object that can be used
* to identify the file.
* @param offset The offset from the beginning of the file from which to start accessing the
* data.
* @param length The amounts of bytes, starting from the offset, that this stream can access
* from the file.
*
* @remark The caller owns the file handle and needs to open it along with keeping it alive
* for the necessary duration. The caller is also responsible for closing it once they are
* done.
*/
RandomAccessFileBodyStream(int fileDescriptor, int64_t offset, int64_t length)
: m_fileDescriptor(fileDescriptor), m_baseOffset(offset), m_length(length), m_offset(0)
{
}
RandomAccessFileBodyStream() : m_fileDescriptor(0), m_baseOffset(0), m_length(0), m_offset(0)
{
}
#elif defined(AZ_PLATFORM_WINDOWS)
/**
* @brief Construct from a file handle.
*
* @param fileHandle A file handle to an already opened file object that can be used to
* identify the file.
* @param offset The offset from the beginning of the file from which to start accessing the
* data.
* @param length The amounts of bytes, starting from the offset, that this stream can access
* from the file.
*
* @remark The caller owns the file handle and needs to open it along with keeping it alive
* for the necessary duration. The caller is also responsible for closing it once they are
* done.
*/
RandomAccessFileBodyStream(HANDLE fileHandle, int64_t offset, int64_t length)
: m_filehandle(fileHandle), m_baseOffset(offset), m_length(length), m_offset(0)
{
}
RandomAccessFileBodyStream() : m_filehandle(NULL), m_baseOffset(0), m_length(0), m_offset(0)
{
}
#endif
// Rewind seeks back to 0
void Rewind() override { this->m_offset = 0; }
int64_t Length() const override { return this->m_length; };
};
} // namespace _internal
/**
* @brief #Azure::Core::IO::BodyStream providing its data from a file.
* @brief A concrete implementation of #Azure::Core::IO::BodyStream used for reading data from a
* file.
*/
class FileBodyStream : public BodyStream {
private:
// immutable
#if defined(AZ_PLATFORM_POSIX)
int m_fd;
#elif defined(AZ_PLATFORM_WINDOWS)
HANDLE m_hFile;
#if defined(AZ_PLATFORM_WINDOWS)
HANDLE m_filehandle;
#elif defined(AZ_PLATFORM_POSIX)
int m_fileDescriptor;
#endif
int64_t m_baseOffset;
int64_t m_length;
// mutable
int64_t m_offset;
std::unique_ptr<_internal::RandomAccessFileBodyStream> m_randomAccessFileBodyStream;
int64_t OnRead(uint8_t* buffer, int64_t count, Azure::Core::Context const& context) override;
public:
#if defined(AZ_PLATFORM_POSIX)
/**
* @brief Construct from a file.
* @brief Construct from a file name.
*
* @param fd File descriptor.
* @param offset Offset in the file to start providing the data from.
* @param length Length of the data, in bytes, to provide.
* @param filename A reference to a file name string used to identify the file, which needs to
* have the necessary file path specified to locate the file.
*
* @remark The #Azure::Core::IO::FileBodyStream owns the file object and is responsible for
* opening and closing the file.
*
* @remark Do not write to the file while it is being used by the stream.
*/
FileBodyStream(int fd, int64_t offset, int64_t length)
: m_fd(fd), m_baseOffset(offset), m_length(length), m_offset(0)
{
}
#elif defined(AZ_PLATFORM_WINDOWS)
FileBodyStream(const std::string& filename);
/**
* @brief Construct from a file.
* @brief Closes the file and cleans up any resources.
*
* @param hFile File handle.
* @param offset Offset in the file to start providing the data from.
* @param length Length of the data, in bytes, to provide.
*/
FileBodyStream(HANDLE hFile, int64_t offset, int64_t length)
: m_hFile(hFile), m_baseOffset(offset), m_length(length), m_offset(0)
{
}
#endif
~FileBodyStream();
// Rewind seek back to 0
void Rewind() override { this->m_offset = 0; }
// Rewind seeks back to 0
void Rewind() override;
int64_t Length() const override { return this->m_length; };
int64_t Length() const override;
};
}}} // namespace Azure::Core::IO

143
sdk/core/azure-core/src/io/body_stream.cpp
View File

@ -5,7 +5,9 @@
#if defined(AZ_PLATFORM_POSIX)
#include <errno.h>
#include <unistd.h>
#include <fcntl.h> // for open and _O_RDONLY
#include <sys/types.h> // for lseek
#include <unistd.h> // for lseek
#elif defined(AZ_PLATFORM_WINDOWS)
#if !defined(WIN32_LEAN_AND_MEAN)
#define WIN32_LEAN_AND_MEAN
@ -20,6 +22,7 @@
#include "azure/core/io/body_stream.hpp"
#include <algorithm>
#include <codecvt>
#include <cstdint>
#include <cstdio>
#include <cstring>
@ -82,54 +85,98 @@ int64_t MemoryBodyStream::OnRead(uint8_t* buffer, int64_t count, Context const&
return copy_length;
}
#if defined(AZ_PLATFORM_POSIX)
int64_t FileBodyStream::OnRead(uint8_t* buffer, int64_t count, Azure::Core::Context const& context)
FileBodyStream::FileBodyStream(const std::string& filename)
{
(void)context;
auto result = pread(
this->m_fd,
buffer,
std::min(count, this->m_length - this->m_offset),
this->m_baseOffset + this->m_offset);
#if defined(AZ_PLATFORM_WINDOWS)
if (result < 0)
try
{
throw std::runtime_error("Reading error. (Code Number: " + std::to_string(errno) + ")");
}
this->m_offset += result;
return result;
}
#elif defined(AZ_PLATFORM_WINDOWS)
int64_t FileBodyStream::OnRead(uint8_t* buffer, int64_t count, Azure::Core::Context const& context)
{
(void)context;
DWORD numberOfBytesRead;
auto o = OVERLAPPED();
o.Offset = static_cast<DWORD>(this->m_baseOffset + this->m_offset);
o.OffsetHigh = static_cast<DWORD>((this->m_baseOffset + this->m_offset) >> 32);
auto result = ReadFile(
this->m_hFile,
buffer,
// at most 4Gb to be read
static_cast<DWORD>(std::min(
static_cast<uint64_t>(0xFFFFFFFFUL),
static_cast<uint64_t>(std::min(count, (this->m_length - this->m_offset))))),
&numberOfBytesRead,
&o);
if (!result)
{
// Check error. of EOF, return bytes read to EOF
auto error = GetLastError();
if (error != ERROR_HANDLE_EOF)
{
throw std::runtime_error("Reading error. (Code Number: " + std::to_string(error) + ")");
}
}
this->m_offset += numberOfBytesRead;
return numberOfBytesRead;
}
#if !defined(WINAPI_PARTITION_DESKTOP) \
|| WINAPI_PARTITION_DESKTOP // See azure/core/platform.hpp for explanation.
m_filehandle = CreateFile(
filename.data(),
GENERIC_READ,
FILE_SHARE_READ,
nullptr,
OPEN_EXISTING,
FILE_FLAG_SEQUENTIAL_SCAN, // Using this as an optimization since we know file access is
// intended to be sequential from beginning to end.
NULL);
#else
m_filehandle = CreateFile2(
std::wstring_convert<std::codecvt_utf8_utf16<wchar_t>>().from_bytes(filename).c_str(),
GENERIC_READ,
FILE_SHARE_READ,
OPEN_EXISTING,
NULL);
#endif
if (m_filehandle == INVALID_HANDLE_VALUE)
{
throw std::runtime_error("Failed to open file for reading. File name: '" + filename + "'");
}
LARGE_INTEGER fileSize;
if (!GetFileSizeEx(m_filehandle, &fileSize))
{
throw std::runtime_error("Failed to get size of file. File name: '" + filename + "'");
}
m_randomAccessFileBodyStream = std::make_unique<_internal::RandomAccessFileBodyStream>(
_internal::RandomAccessFileBodyStream(m_filehandle, 0, fileSize.QuadPart));
}
catch (std::exception&)
{
CloseHandle(m_filehandle);
throw;
}
#elif defined(AZ_PLATFORM_POSIX)
try
{
m_fileDescriptor = open(filename.data(), O_RDONLY);
if (m_fileDescriptor == -1)
{
throw std::runtime_error("Failed to open file for reading. File name: '" + filename + "'");
}
int64_t fileSize = lseek(m_fileDescriptor, 0, SEEK_END);
if (fileSize == -1)
{
throw std::runtime_error("Failed to get size of file. File name: '" + filename + "'");
}
m_randomAccessFileBodyStream = std::make_unique<_internal::RandomAccessFileBodyStream>(
_internal::RandomAccessFileBodyStream(m_fileDescriptor, 0, fileSize));
}
catch (std::exception&)
{
close(m_fileDescriptor);
throw;
}
#endif
}
FileBodyStream::~FileBodyStream()
{
#if defined(AZ_PLATFORM_WINDOWS)
if (m_filehandle)
{
CloseHandle(m_filehandle);
m_filehandle = NULL;
}
#elif defined(AZ_PLATFORM_POSIX)
if (m_fileDescriptor)
{
close(m_fileDescriptor);
m_fileDescriptor = 0;
}
#endif
}
int64_t FileBodyStream::OnRead(uint8_t* buffer, int64_t count, Azure::Core::Context const& context)
{
return m_randomAccessFileBodyStream->Read(buffer, count, context);
}
void FileBodyStream::Rewind() { m_randomAccessFileBodyStream->Rewind(); }
int64_t FileBodyStream::Length() const { return m_randomAccessFileBodyStream->Length(); }

78
sdk/core/azure-core/src/io/random_access_file_body_stream.cpp Normal file
View File

@ -0,0 +1,78 @@
// Copyright (c) Microsoft Corporation. All rights reserved.
// SPDX-License-Identifier: MIT
#include "azure/core/platform.hpp"
#if defined(AZ_PLATFORM_POSIX)
#include <errno.h>
#include <unistd.h>
#elif defined(AZ_PLATFORM_WINDOWS)
#if !defined(WIN32_LEAN_AND_MEAN)
#define WIN32_LEAN_AND_MEAN
#endif
#if !defined(NOMINMAX)
#define NOMINMAX
#endif
#include <windows.h>
#endif
#include "azure/core/context.hpp"
#include "azure/core/io/body_stream.hpp"
#include <cstdint>
#include <stdexcept>
#include <string>
using Azure::Core::Context;
using namespace Azure::Core::IO::_internal;
int64_t RandomAccessFileBodyStream::OnRead(
uint8_t* buffer,
int64_t count,
Azure::Core::Context const&)
{
#if defined(AZ_PLATFORM_POSIX)
auto numberOfBytesRead = pread(
this->m_fileDescriptor,
buffer,
std::min(count, this->m_length - this->m_offset),
this->m_baseOffset + this->m_offset);
if (numberOfBytesRead < 0)
{
throw std::runtime_error("Reading error. (Code Number: " + std::to_string(errno) + ")");
}
#elif defined(AZ_PLATFORM_WINDOWS)
DWORD numberOfBytesRead;
auto o = OVERLAPPED();
o.Offset = static_cast<DWORD>(this->m_baseOffset + this->m_offset);
o.OffsetHigh = static_cast<DWORD>((this->m_baseOffset + this->m_offset) >> 32);
auto result = ReadFile(
this->m_filehandle,
buffer,
// at most 4Gb to be read
static_cast<DWORD>(std::min(
static_cast<uint64_t>(0xFFFFFFFFUL),
static_cast<uint64_t>(std::min(count, (this->m_length - this->m_offset))))),
&numberOfBytesRead,
&o);
if (!result)
{
// Check error. If EOF, return bytes read to EOF.
auto error = GetLastError();
if (error != ERROR_HANDLE_EOF)
{
throw std::runtime_error("Reading error. (Code Number: " + std::to_string(error) + ")");
}
}
#endif
this->m_offset += numberOfBytesRead;
return numberOfBytesRead;
}

38
sdk/core/azure-core/test/ut/bodystream.cpp
View File

@ -39,26 +39,40 @@ TEST(BodyStream, Rewind)
#if defined(AZ_PLATFORM_POSIX)
testDataPath.append("/fileData");
int f = open(testDataPath.data(), O_RDONLY);
EXPECT_GE(f, 0);
#elif defined(AZ_PLATFORM_WINDOWS)
testDataPath.append("\\fileData");
HANDLE f = CreateFile(
testDataPath.data(),
GENERIC_READ,
FILE_SHARE_READ,
NULL,
OPEN_EXISTING,
FILE_FLAG_SEQUENTIAL_SCAN,
NULL);
EXPECT_NE(f, INVALID_HANDLE_VALUE);
#else
#error "Unknown platform"
#endif
auto fileBodyStream = Azure::Core::IO::FileBodyStream(f, 0, 0);
Azure::Core::IO::FileBodyStream fileBodyStream(testDataPath);
EXPECT_NO_THROW(fileBodyStream.Rewind());
std::vector<uint8_t> data = {1, 2, 3, 4};
Azure::Core::IO::MemoryBodyStream ms(data);
EXPECT_NO_THROW(ms.Rewind());
}
TEST(FileBodyStream, BadInput)
{
EXPECT_THROW(Azure::Core::IO::FileBodyStream(""), std::runtime_error);
EXPECT_THROW(Azure::Core::IO::FileBodyStream("FileNotFound"), std::runtime_error);
}
constexpr int64_t FileSize = 1024 * 100;
TEST(FileBodyStream, Length)
{
std::string testDataPath(AZURE_TEST_DATA_PATH);
testDataPath.append("/fileData");
Azure::Core::IO::FileBodyStream stream(testDataPath);
EXPECT_EQ(stream.Length(), FileSize);
auto readResult = Azure::Core::IO::BodyStream::ReadToEnd(
stream, Azure::Core::Context::GetApplicationContext());
EXPECT_EQ(readResult.size(), FileSize);
stream.Rewind();
EXPECT_EQ(stream.Length(), FileSize);
}

43
sdk/core/azure-core/test/ut/transport_adapter_base.cpp
View File

@ -440,24 +440,13 @@ namespace Azure { namespace Core { namespace Test {
#if defined(AZ_PLATFORM_POSIX)
testDataPath.append("/fileData");
int f = open(testDataPath.data(), O_RDONLY);
EXPECT_GE(f, 0);
#elif defined(AZ_PLATFORM_WINDOWS)
testDataPath.append("\\fileData");
HANDLE f = CreateFile(
testDataPath.data(),
GENERIC_READ,
FILE_SHARE_READ,
NULL,
OPEN_EXISTING,
FILE_FLAG_SEQUENTIAL_SCAN,
NULL);
EXPECT_NE(f, INVALID_HANDLE_VALUE);
#else
#error "Unknown platform"
#endif
auto requestBodyStream
= Azure::Core::IO::FileBodyStream(f, 0, Azure::Core::Test::Datails::FileSize);
Azure::Core::IO::FileBodyStream requestBodyStream(testDataPath);
auto request = Azure::Core::Http::Request(
Azure::Core::Http::HttpMethod::Put, host, &requestBodyStream, true);
// Make transport adapter to read all stream content for uploading instead of chunks
@ -478,25 +467,13 @@ namespace Azure { namespace Core { namespace Test {
#if defined(AZ_PLATFORM_POSIX)
testDataPath.append("/fileData");
int f = open(testDataPath.data(), O_RDONLY);
EXPECT_GE(f, 0);
#elif defined(AZ_PLATFORM_WINDOWS)
testDataPath.append("\\fileData");
HANDLE f = CreateFile(
testDataPath.data(),
GENERIC_READ,
FILE_SHARE_READ,
NULL,
OPEN_EXISTING,
FILE_FLAG_SEQUENTIAL_SCAN,
NULL);
EXPECT_NE(f, INVALID_HANDLE_VALUE);
#else
#error "Unknown platform"
#endif
auto requestBodyStream
= Azure::Core::IO::FileBodyStream(f, 0, Azure::Core::Test::Datails::FileSize);
Azure::Core::IO::FileBodyStream requestBodyStream(testDataPath);
auto request = Azure::Core::Http::Request(
Azure::Core::Http::HttpMethod::Put, host, &requestBodyStream, true);
// Make transport adapter to read default chunk size
@ -516,25 +493,13 @@ namespace Azure { namespace Core { namespace Test {
#if defined(AZ_PLATFORM_POSIX)
testDataPath.append("/fileData");
int f = open(testDataPath.data(), O_RDONLY);
EXPECT_GE(f, 0);
#elif defined(AZ_PLATFORM_WINDOWS)
testDataPath.append("\\fileData");
HANDLE f = CreateFile(
testDataPath.data(),
GENERIC_READ,
FILE_SHARE_READ,
NULL,
OPEN_EXISTING,
FILE_FLAG_SEQUENTIAL_SCAN,
NULL);
EXPECT_NE(f, INVALID_HANDLE_VALUE);
#else
#error "Unknown platform"
#endif
auto requestBodyStream
= Azure::Core::IO::FileBodyStream(f, 0, Azure::Core::Test::Datails::FileSize);
Azure::Core::IO::FileBodyStream requestBodyStream(testDataPath);
auto request = Azure::Core::Http::Request(
Azure::Core::Http::HttpMethod::Put, host, &requestBodyStream, true);
// Make transport adapter to read more than file size (5Mb)

30
sdk/storage/azure-storage-blobs/src/block_blob_client.cpp
View File

@ -3,6 +3,7 @@
#include "azure/storage/blobs/block_blob_client.hpp"
#include <azure/core/io/body_stream.hpp>
#include <azure/storage/common/concurrent_transfer.hpp>
#include <azure/storage/common/constants.hpp>
#include <azure/storage/common/crypt.hpp>
@ -173,19 +174,17 @@ namespace Azure { namespace Storage { namespace Blobs {
{
constexpr int64_t MaxStageBlockSize = 4000 * 1024 * 1024ULL;
Storage::_detail::FileReader fileReader(fileName);
int64_t chunkSize = std::min(MaxStageBlockSize, options.TransferOptions.ChunkSize);
if (fileReader.GetFileSize() <= options.TransferOptions.SingleUploadThreshold)
{
Azure::Core::IO::FileBodyStream contentStream(
fileReader.GetHandle(), 0, fileReader.GetFileSize());
UploadBlockBlobOptions uploadBlockBlobOptions;
uploadBlockBlobOptions.HttpHeaders = options.HttpHeaders;
uploadBlockBlobOptions.Metadata = options.Metadata;
uploadBlockBlobOptions.Tier = options.Tier;
return Upload(&contentStream, uploadBlockBlobOptions, context);
Azure::Core::IO::FileBodyStream contentStream(fileName);
if (contentStream.Length() <= options.TransferOptions.SingleUploadThreshold)
{
UploadBlockBlobOptions uploadBlockBlobOptions;
uploadBlockBlobOptions.HttpHeaders = options.HttpHeaders;
uploadBlockBlobOptions.Metadata = options.Metadata;
uploadBlockBlobOptions.Tier = options.Tier;
return Upload(&contentStream, uploadBlockBlobOptions, context);
}
}
std::vector<std::string> blockIds;
@ -196,8 +195,11 @@ namespace Azure { namespace Storage { namespace Blobs {
return Azure::Core::Base64Encode(std::vector<uint8_t>(blockId.begin(), blockId.end()));
};
Storage::_detail::FileReader fileReader(fileName);
auto uploadBlockFunc = [&](int64_t offset, int64_t length, int64_t chunkId, int64_t numChunks) {
Azure::Core::IO::FileBodyStream contentStream(fileReader.GetHandle(), offset, length);
Azure::Core::IO::_internal::RandomAccessFileBodyStream contentStream(
fileReader.GetHandle(), offset, length);
StageBlockOptions chunkOptions;
auto blockInfo = StageBlock(getBlockId(chunkId), &contentStream, chunkOptions, context);
if (chunkId == numChunks - 1)
@ -206,6 +208,8 @@ namespace Azure { namespace Storage { namespace Blobs {
}
};
int64_t chunkSize = std::min(MaxStageBlockSize, options.TransferOptions.ChunkSize);
Storage::_detail::ConcurrentTransfer(
0,
fileReader.GetFileSize(),

4
sdk/storage/azure-storage-files-shares/src/share_file_client.cpp
View File

@ -6,6 +6,7 @@
#include <azure/core/credentials.hpp>
#include <azure/core/http/policy.hpp>
#include <azure/core/internal/null_body_stream.hpp>
#include <azure/core/io/body_stream.hpp>
#include <azure/storage/common/concurrent_transfer.hpp>
#include <azure/storage/common/constants.hpp>
#include <azure/storage/common/crypt.hpp>
@ -1024,7 +1025,8 @@ namespace Azure { namespace Storage { namespace Files { namespace Shares {
auto uploadPageFunc = [&](int64_t offset, int64_t length, int64_t chunkId, int64_t numChunks) {
(void)chunkId;
(void)numChunks;
Azure::Core::IO::FileBodyStream contentStream(fileReader.GetHandle(), offset, length);
Azure::Core::IO::_internal::RandomAccessFileBodyStream contentStream(
fileReader.GetHandle(), offset, length);
UploadShareFileRangeOptions uploadRangeOptions;
UploadRange(offset, &contentStream, uploadRangeOptions, context);
};