db/pgbackrest
1
0
Fork 0
mirror of https://github.com/pgbackrest/pgbackrest.git synced 2026-06-20 01:17:49 +02:00
Code Issues Releases Activity

Optimize S3/GCS/Azure chunk size for small files.

Browse Source 
The prior code allocated the entire chunk buffer when the file was opened. However, in practice many files are smaller than the chunk buffer, especially in the main process.

Instead grow the chunk buffer as data comes in to save memory when smaller files are being processed. This adds some overhead for reallocations but modern processors do this very efficiently so it should not be significant compared to the cost of compressing, encrypting, and transferring files. Even so, the growth is fairly aggressive when the input buffers are full so only one or two reallocation are required to get to the default chunk size.
This commit is contained in:
David Steele
2025-11-25 17:28:20 +05:30
committed by GitHub
parent 86554faa70
commit 28bfa20ba3
10 changed files with 111 additions and 129 deletions
Download Patch File Download Diff File Expand all files Collapse all files
doc/xml/release/2020s/2026/2.58.0.xml
+11
View File
@@ -31,6 +31,17 @@
<p>Dynamically size <proper>S3</proper>/<proper>GCS</proper>/<proper>Azure</proper> chunks for large uploads.</p>
</release-item>
<release-item>
<github-pull-request id="2705"/>
<release-item-contributor-list>
<release-item-contributor id="david.steele"/>
<release-item-reviewer id="douglas.j.hunley"/>
</release-item-contributor-list>
<p>Optimize <proper>S3</proper>/<proper>GCS</proper>/<proper>Azure</proper> chunk size for small files.</p>
</release-item>
<release-item>
<github-pull-request id="2692"/>
src/storage/azure/write.c
+9 -33
View File
@@ -61,31 +61,6 @@ Macros for function logging
#define FUNCTION_LOG_STORAGE_WRITE_AZURE_FORMAT(value, buffer, bufferSize) \
objNameToLog(value, "StorageWriteAzure", buffer, bufferSize)
/***********************************************************************************************************************************
Open the file
***********************************************************************************************************************************/
static void
storageWriteAzureOpen(THIS_VOID)
{
THIS(StorageWriteAzure);
FUNCTION_LOG_BEGIN(logLevelTrace);
FUNCTION_LOG_PARAM(STORAGE_WRITE_AZURE, this);
FUNCTION_LOG_END();
ASSERT(this != NULL);
ASSERT(this->blockBuffer == NULL);
// Allocate the block buffer
MEM_CONTEXT_OBJ_BEGIN(this)
{
this->blockBuffer = bufNew(this->blockSize);
}
MEM_CONTEXT_OBJ_END();
FUNCTION_LOG_RETURN_VOID();
}
/***********************************************************************************************************************************
Flush bytes to upload block
***********************************************************************************************************************************/
@@ -176,11 +151,13 @@ storageWriteAzure(THIS_VOID, const Buffer *const buffer)
ASSERT(this != NULL);
ASSERT(buffer != NULL);
ASSERT(this->blockBuffer != NULL);
size_t bytesTotal = 0;
// Resize chunk buffer
storageWriteChunkBufferResize(buffer, this->blockBuffer, this->blockSize);
// Continue until the write buffer has been exhausted
size_t bytesTotal = 0;
do
{
// Copy as many bytes as possible into the block buffer
@@ -192,15 +169,14 @@ storageWriteAzure(THIS_VOID, const Buffer *const buffer)
bytesTotal += bytesNext;
// If the block buffer is full then write it
if (bufRemains(this->blockBuffer) == 0)
if (bufUsed(this->blockBuffer) == this->blockSize)
{
storageWriteAzureBlockAsync(this);
bufUsedZero(this->blockBuffer);
bufResize(
this->blockBuffer,
this->blockSize =
storageWriteChunkSize(
this->blockSize, STORAGE_AZURE_SPLIT_DEFAULT, STORAGE_AZURE_SPLIT_MAX, strLstSize(this->blockIdList)));
this->blockSize, STORAGE_AZURE_SPLIT_DEFAULT, STORAGE_AZURE_SPLIT_MAX, strLstSize(this->blockIdList));
}
}
while (bytesTotal != bufUsed(buffer));
@@ -292,6 +268,7 @@ storageWriteAzureNew(StorageAzure *const storage, const String *const name, cons
.storage = storage,
.fileId = fileId,
.blockSize = blockSize,
.blockBuffer = bufNew(0),
.interface = (StorageWriteInterface)
{
@@ -306,7 +283,6 @@ storageWriteAzureNew(StorageAzure *const storage, const String *const name, cons
.ioInterface = (IoWriteInterface)
{
.close = storageWriteAzureClose,
.open = storageWriteAzureOpen,
.write = storageWriteAzure,
},
},
src/storage/gcs/write.c
+10 -34
View File
@@ -52,32 +52,6 @@ Macros for function logging
#define FUNCTION_LOG_STORAGE_WRITE_GCS_FORMAT(value, buffer, bufferSize) \
objNameToLog(value, "StorageWriteGcs", buffer, bufferSize)
/***********************************************************************************************************************************
Open the file
***********************************************************************************************************************************/
static void
storageWriteGcsOpen(THIS_VOID)
{
THIS(StorageWriteGcs);
FUNCTION_LOG_BEGIN(logLevelTrace);
FUNCTION_LOG_PARAM(STORAGE_WRITE_GCS, this);
FUNCTION_LOG_END();
ASSERT(this != NULL);
ASSERT(this->chunkBuffer == NULL);
// Allocate the chunk buffer
MEM_CONTEXT_OBJ_BEGIN(this)
{
this->chunkBuffer = bufNew(this->chunkSize);
this->md5hash = cryptoHashNew(hashTypeMd5);
}
MEM_CONTEXT_OBJ_END();
FUNCTION_LOG_RETURN_VOID();
}
/***********************************************************************************************************************************
Verify upload
***********************************************************************************************************************************/
@@ -243,11 +217,13 @@ storageWriteGcs(THIS_VOID, const Buffer *const buffer)
ASSERT(this != NULL);
ASSERT(buffer != NULL);
ASSERT(this->chunkBuffer != NULL);
size_t bytesTotal = 0;
// Resize chunk buffer
storageWriteChunkBufferResize(buffer, this->chunkBuffer, this->chunkSize);
// Continue until the write buffer has been exhausted
size_t bytesTotal = 0;
do
{
// Copy as many bytes as possible into the chunk buffer
@@ -260,14 +236,13 @@ storageWriteGcs(THIS_VOID, const Buffer *const buffer)
// If the chunk buffer is full then write it. It is possible that this is the last chunk and it would be better to wait, but
// the chances of that are quite small so in general it is better to write now so there is less to write later.
if (bufRemains(this->chunkBuffer) == 0)
if (bufUsed(this->chunkBuffer) == this->chunkSize)
{
storageWriteGcsBlockAsync(this, false);
bufUsedZero(this->chunkBuffer);
bufResize(
this->chunkBuffer,
storageWriteChunkSize(this->chunkSize, STORAGE_GCS_SPLIT_DEFAULT, STORAGE_GCS_SPLIT_MAX, ++this->chunkTotal));
this->chunkSize =
storageWriteChunkSize(this->chunkSize, STORAGE_GCS_SPLIT_DEFAULT, STORAGE_GCS_SPLIT_MAX, ++this->chunkTotal);
}
}
while (bytesTotal != bufUsed(buffer));
@@ -351,6 +326,8 @@ storageWriteGcsNew(StorageGcs *const storage, const String *const name, const si
{
.storage = storage,
.chunkSize = chunkSize,
.chunkBuffer = bufNew(0),
.md5hash = cryptoHashNew(hashTypeMd5),
.tag = tag,
.interface = (StorageWriteInterface)
@@ -366,7 +343,6 @@ storageWriteGcsNew(StorageGcs *const storage, const String *const name, const si
.ioInterface = (IoWriteInterface)
{
.close = storageWriteGcsClose,
.open = storageWriteGcsOpen,
.write = storageWriteGcs,
},
},
src/storage/s3/write.c
+9 -34
View File
@@ -55,31 +55,6 @@ Macros for function logging
#define FUNCTION_LOG_STORAGE_WRITE_S3_FORMAT(value, buffer, bufferSize) \
objNameToLog(value, "StorageWriteS3", buffer, bufferSize)
/***********************************************************************************************************************************
Open the file
***********************************************************************************************************************************/
static void
storageWriteS3Open(THIS_VOID)
{
THIS(StorageWriteS3);
FUNCTION_LOG_BEGIN(logLevelTrace);
FUNCTION_LOG_PARAM(STORAGE_WRITE_S3, this);
FUNCTION_LOG_END();
ASSERT(this != NULL);
ASSERT(this->partBuffer == NULL);
// Allocate the part buffer
MEM_CONTEXT_OBJ_BEGIN(this)
{
this->partBuffer = bufNew(this->partSize);
}
MEM_CONTEXT_OBJ_END();
FUNCTION_LOG_RETURN_VOID();
}
/***********************************************************************************************************************************
Flush bytes to upload part
***********************************************************************************************************************************/
@@ -176,12 +151,14 @@ storageWriteS3(THIS_VOID, const Buffer *const buffer)
FUNCTION_LOG_END();
ASSERT(this != NULL);
ASSERT(this->partBuffer != NULL);
ASSERT(buffer != NULL);
size_t bytesTotal = 0;
// Resize chunk buffer
storageWriteChunkBufferResize(buffer, this->partBuffer, this->partSize);
// Continue until the write buffer has been exhausted
size_t bytesTotal = 0;
do
{
// Copy as many bytes as possible into the part buffer
@@ -193,15 +170,13 @@ storageWriteS3(THIS_VOID, const Buffer *const buffer)
bytesTotal += bytesNext;
// If the part buffer is full then write it
if (bufRemains(this->partBuffer) == 0)
if (bufUsed(this->partBuffer) == this->partSize)
{
storageWriteS3PartAsync(this);
bufUsedZero(this->partBuffer);
bufResize(
this->partBuffer,
storageWriteChunkSize(
this->partSize, STORAGE_S3_SPLIT_DEFAULT, STORAGE_S3_SPLIT_MAX, strLstSize(this->uploadPartList)));
this->partSize = storageWriteChunkSize(
this->partSize, STORAGE_S3_SPLIT_DEFAULT, STORAGE_S3_SPLIT_MAX, strLstSize(this->uploadPartList));
}
}
while (bytesTotal != bufUsed(buffer));
@@ -298,6 +273,7 @@ storageWriteS3New(StorageS3 *const storage, const String *const name, const size
{
.storage = storage,
.partSize = partSize,
.partBuffer = bufNew(0),
.interface = (StorageWriteInterface)
{
@@ -312,7 +288,6 @@ storageWriteS3New(StorageS3 *const storage, const String *const name, const size
.ioInterface = (IoWriteInterface)
{
.close = storageWriteS3Close,
.open = storageWriteS3Open,
.write = storageWriteS3,
},
},
src/storage/write.c
+46
View File
@@ -105,6 +105,52 @@ storageWriteChunkSize(
FUNCTION_TEST_RETURN(SIZE, (size_t)result);
}
/**********************************************************************************************************************************/
FN_EXTERN void
storageWriteChunkBufferResize(const Buffer *const input, Buffer *const chunk, const size_t chunkSizeMax)
{
FUNCTION_TEST_BEGIN();
FUNCTION_TEST_PARAM(BUFFER, input);
FUNCTION_TEST_PARAM(BUFFER, chunk);
FUNCTION_TEST_PARAM(SIZE, chunkSizeMax);
FUNCTION_TEST_END();
ASSERT(input != NULL);
ASSERT(chunk != NULL);
ASSERT(chunkSizeMax > 0);
// Resize chunk buffer if it is less than max chunk size
size_t chunkSize = bufSize(chunk);
if (chunkSize < chunkSizeMax)
{
// If the input buffer is full there is very likely more data so increase size of the chunk buffer aggressively
if (bufFull(input))
{
// If this is the first write set chunk size equal to double the input buffer size
if (chunkSize == 0)
{
chunkSize = bufSize(input) * 2;
}
// Else double chunk size so as not to resize the chunk buffer too many times since prior data must be copied
else
chunkSize *= 2;
}
// Else no more writes are expected so allocate only what is needed
else
chunkSize += bufUsed(input) - bufRemains(chunk);
// Chunk size cannot be larger than max chunk size
if (chunkSize > chunkSizeMax)
chunkSize = chunkSizeMax;
// Resize the chunk buffer
bufResize(chunk, chunkSize);
}
FUNCTION_TEST_RETURN_VOID();
}
/**********************************************************************************************************************************/
FN_EXTERN void
storageWriteToLog(const StorageWrite *const this, StringStatic *const debugLog)
src/storage/write.h
+3
View File
@@ -116,6 +116,9 @@ Helper functions
FN_EXTERN size_t storageWriteChunkSize(
size_t chunkSizeDefault, unsigned int defaultSplit, unsigned int maxSplit, unsigned int chunkIdx);
// Resize chunk buffer for multipart upload to an object store
FN_EXTERN void storageWriteChunkBufferResize(const Buffer *input, Buffer *const chunk, size_t chunkSizeMax);
/***********************************************************************************************************************************
Macros for function logging
***********************************************************************************************************************************/
test/define.yaml
+1 -1
View File
@@ -620,11 +620,11 @@ unit:
- storage/s3/storage
- storage/s3/write
- storage/read
- storage/write
- storage/helper
include:
- storage/storage
- storage/write
# ----------------------------------------------------------------------------------------------------------------------------
- name: sftp
test/src/module/storage/azureTest.c
+6 -7
View File
@@ -707,11 +707,11 @@ testRun(void)
testRequestP(
service, HTTP_VERB_PUT, "/file.txt?blockid=0AAAAAAACCCCCCCDx0000000&comp=block",
.content = "12345678901234567");
.content = "1234567890123456");
testResponseP(service);
testRequestP(
service, HTTP_VERB_PUT, "/file.txt?blockid=0AAAAAAACCCCCCCDx0000001&comp=block", .content = "890");
service, HTTP_VERB_PUT, "/file.txt?blockid=0AAAAAAACCCCCCCDx0000001&comp=block", .content = "7890");
testResponseP(service);
testRequestP(
@@ -725,23 +725,22 @@ testRun(void)
testResponseP(service);
// Check that block size is updated during write
ioBufferSizeSet(9);
ioBufferSizeSet(6);
TEST_ASSIGN(write, storageNewWriteP(storage, STRDEF("file.txt")), "new write");
ioWriteOpen(storageWriteIo(write));
ioWrite(storageWriteIo(write), BUFSTRDEF("123456789012345678"));
TEST_RESULT_VOID(
bufResize(((StorageWriteAzure *)ioWriteDriver(storageWriteIo(write)))->blockBuffer, 17),
"resize part buffer to 17");
ioWrite(storageWriteIo(write), BUFSTRDEF("123456789012345678"));
ioWrite(storageWriteIo(write), BUFSTRDEF("90"));
TEST_RESULT_UINT(
bufSize(((StorageWriteAzure *)ioWriteDriver(storageWriteIo(write)))->blockBuffer), 16,
((StorageWriteAzure *)ioWriteDriver(storageWriteIo(write)))->blockSize, 16,
"part buffer reset to 16 (default)");
ioWrite(storageWriteIo(write), BUFSTRDEF("90"));
ioWriteClose(storageWriteIo(write));
ioBufferSizeSet(ioBufferSizeDefault);
test/src/module/storage/gcsTest.c
+10 -12
View File
@@ -711,38 +711,36 @@ testRun(void)
testRequestP(
service, HTTP_VERB_PUT, .upload = true, .noAuth = true,
.query = "name=file.txt&uploadType=resumable&upload_id=ulid2", .contentRange = "0-16/*",
.content = "12345678901234567");
.query = "name=file.txt&uploadType=resumable&upload_id=ulid2", .contentRange = "0-15/*",
.content = "1234567890123456");
testResponseP(service, .code = 503);
testRequestP(
service, HTTP_VERB_PUT, .upload = true, .noAuth = true,
.query = "name=file.txt&uploadType=resumable&upload_id=ulid2", .contentRange = "0-16/*",
.content = "12345678901234567");
.query = "name=file.txt&uploadType=resumable&upload_id=ulid2", .contentRange = "0-15/*",
.content = "1234567890123456");
testResponseP(service, .code = 308);
testRequestP(
service, HTTP_VERB_PUT, .upload = true, .noAuth = true,
.query = "fields=md5Hash%2Csize&name=file.txt&uploadType=resumable&upload_id=ulid2", .contentRange = "17-19/20",
.content = "890");
.query = "fields=md5Hash%2Csize&name=file.txt&uploadType=resumable&upload_id=ulid2", .contentRange = "16-19/20",
.content = "7890");
testResponseP(service, .content = "{\"md5Hash\":\"/YXmLZvrRUKHcexohBiycQ==\",\"size\":\"20\"}");
// Check that chunk size is updated during write
ioBufferSizeSet(9);
ioBufferSizeSet(6);
TEST_ASSIGN(write, storageNewWriteP(storage, STRDEF("file.txt")), "new write");
ioWriteOpen(storageWriteIo(write));
ioWrite(storageWriteIo(write), BUFSTRDEF("123456789012345678"));
TEST_RESULT_VOID(
bufResize(((StorageWriteGcs *)ioWriteDriver(storageWriteIo(write)))->chunkBuffer, 17),
"resize part buffer to 17");
ioWrite(storageWriteIo(write), BUFSTRDEF("123456789012345678"));
ioWrite(storageWriteIo(write), BUFSTRDEF("90"));
TEST_RESULT_UINT(
bufSize(((StorageWriteGcs *)ioWriteDriver(storageWriteIo(write)))->chunkBuffer), 16,
"part buffer reset to 16 (default)");
ioWrite(storageWriteIo(write), BUFSTRDEF("90"));
((StorageWriteGcs *)ioWriteDriver(storageWriteIo(write)))->chunkSize, 16, "part buffer reset to 16 (default)");
ioWriteClose(storageWriteIo(write));
ioBufferSizeSet(ioBufferSizeDefault);
test/src/module/storage/s3Test.c
+6 -8
View File
@@ -1046,12 +1046,12 @@ testRun(void)
"</InitiateMultipartUploadResult>");
testRequestP(
service, s3, HTTP_VERB_PUT, "/file.txt?partNumber=1&uploadId=RR55", .content = "12345678901234567",
service, s3, HTTP_VERB_PUT, "/file.txt?partNumber=1&uploadId=RR55", .content = "1234567890123456",
.sseC = "rA1P");
testResponseP(service, .header = "etag:RR551");
testRequestP(
service, s3, HTTP_VERB_PUT, "/file.txt?partNumber=2&uploadId=RR55", .content = "890",
service, s3, HTTP_VERB_PUT, "/file.txt?partNumber=2&uploadId=RR55", .content = "7890",
.sseC = "rA1P");
testResponseP(service, .header = "eTag:RR552");
@@ -1070,22 +1070,20 @@ testRun(void)
"<CompleteMultipartUploadResult><ETag>XXX</ETag></CompleteMultipartUploadResult>");
// Check that block size is updated during write
ioBufferSizeSet(9);
ioBufferSizeSet(6);
TEST_ASSIGN(write, storageNewWriteP(s3, STRDEF("file.txt")), "new write");
ioWriteOpen(storageWriteIo(write));
ioWrite(storageWriteIo(write), BUFSTRDEF("123456789012345678"));
TEST_RESULT_VOID(
bufResize(((StorageWriteS3 *)ioWriteDriver(storageWriteIo(write)))->partBuffer, 17),
"resize part buffer to 17");
ioWrite(storageWriteIo(write), BUFSTRDEF("123456789012345678"));
ioWrite(storageWriteIo(write), BUFSTRDEF("90"));
TEST_RESULT_UINT(
bufSize(((StorageWriteS3 *)ioWriteDriver(storageWriteIo(write)))->partBuffer), 16,
"part buffer reset to 16 (default)");
ioWrite(storageWriteIo(write), BUFSTRDEF("90"));
((StorageWriteS3 *)ioWriteDriver(storageWriteIo(write)))->partSize, 16, "part buffer reset to 16 (default)");
ioWriteClose(storageWriteIo(write));
ioBufferSizeSet(ioBufferSizeDefault);