pgbackrest.git / summary / log / commit / refs
commit 2d5cbfc73142fa99df609aa23b64db5342e73c88
Author: David Steele <david@pgmasters.net>
Date: Mon Aug 09 13:28:17 2021 +0000
Fix memory leaks introduced by the binary protocol in 6a1c0337.
Additional buffers were being allocated for the protocol messages but not being freed.
Most of the allocations were fairly harness, but storageRemoteOpenReadProtocol() and storageWriteRemote() were problematic because they were allocating (but not freeing) buffers equal to the transfer size of the file. Depending on compression, this could be a lot of memory. Though the memory was freed after each file transfer the aggregate of memory used during parallel processing could overwhelm systems with constrained memory.
Also allocate larger initial buffers in storageRemoteOpenReadProtocol() and storageWriteRemote() so a reallocation is not needed.
doc/xml/release.xml | 25 +++++++++--------
src/protocol/client.c | 62 ++++++++++++++++++++++++-------------------
src/protocol/server.c | 62 ++++++++++++++++++++++++++-----------------
src/storage/remote/protocol.c | 10 ++++---
src/storage/remote/write.c | 20 ++++++++++----
5 files changed, 108 insertions(+), 71 deletions(-)
diff --git a/doc/xml/release.xml b/doc/xml/release.xml
index 563079152..fc889c7fd 100644
--- a/doc/xml/release.xml
+++ b/doc/xml/release.xml
@@ -68,6 +68,20 @@
</release-bug-list>
<release-improvement-list>
+ <release-item>
+ <commit subject="Binary protocol.">
+ <github-pull-request id="1424"/>
+ </commit>
+ <commit subject="Fix memory leaks introduced by the binary protocol in 6a1c0337."/>
+
+ <release-item-contributor-list>
+ <release-item-contributor id="david.steele"/>
+ <release-item-reviewer id="cynthia.shang"/>
+ </release-item-contributor-list>
+
+ <p>Binary protocol.</p>
+ </release-item>
+
<release-item>
<github-issue id="1445"/>
<github-pull-request id="1454"/>
@@ -122,17 +136,6 @@
</release-improvement-list>
<release-development-list>
- <release-item>
- <github-pull-request id="1424"/>
-
- <release-item-contributor-list>
- <release-item-contributor id="david.steele"/>
- <release-item-reviewer id="cynthia.shang"/>
- </release-item-contributor-list>
-
- <p>Binary protocol.</p>
- </release-item>
-
<release-item>
<github-pull-request id="1421"/>
diff --git a/src/protocol/client.c b/src/protocol/client.c
index 961b9d3d5..cebb6d13b 100644
--- a/src/protocol/client.c
+++ b/src/protocol/client.c
@@ -147,19 +147,23 @@ protocolClientDataPut(ProtocolClient *const this, PackWrite *const data)
ASSERT(this != NULL);
- // End the pack
- if (data != NULL)
- pckWriteEndP(data);
-
- // Write the data
- PackWrite *dataMessage = pckWriteNew(this->write);
- pckWriteU32P(dataMessage, protocolMessageTypeData, .defaultWrite = true);
- pckWritePackP(dataMessage, data);
- pckWriteEndP(dataMessage);
-
- // Flush when there is no more data to put
- if (data == NULL)
- ioWriteFlush(this->write);
+ MEM_CONTEXT_TEMP_BEGIN()
+ {
+ // End the pack
+ if (data != NULL)
+ pckWriteEndP(data);
+
+ // Write the data
+ PackWrite *dataMessage = pckWriteNew(this->write);
+ pckWriteU32P(dataMessage, protocolMessageTypeData, .defaultWrite = true);
+ pckWritePackP(dataMessage, data);
+ pckWriteEndP(dataMessage);
+
+ // Flush when there is no more data to put
+ if (data == NULL)
+ ioWriteFlush(this->write);
+ }
+ MEM_CONTEXT_TEMP_END();
FUNCTION_LOG_RETURN_VOID();
}
@@ -175,26 +179,30 @@ protocolClientError(ProtocolClient *const this, const ProtocolMessageType type,
FUNCTION_LOG_PARAM(PACK_READ, error);
FUNCTION_LOG_END();
- if (type == protocolMessageTypeError)
+ MEM_CONTEXT_TEMP_BEGIN()
{
- const ErrorType *type = errorTypeFromCode(pckReadI32P(error));
- String *const message = strNewFmt("%s: %s", strZ(this->errorPrefix), strZ(pckReadStrP(error)));
- const String *const stack = pckReadStrP(error);
- pckReadEndP(error);
+ if (type == protocolMessageTypeError)
+ {
+ const ErrorType *type = errorTypeFromCode(pckReadI32P(error));
+ String *const message = strNewFmt("%s: %s", strZ(this->errorPrefix), strZ(pckReadStrP(error)));
+ const String *const stack = pckReadStrP(error);
+ pckReadEndP(error);
- CHECK(message != NULL);
+ CHECK(message != NULL);
- // Add stack trace if the error is an assertion or debug-level logging is enabled
- if (type == &AssertError || logAny(logLevelDebug))
- {
- CHECK(stack != NULL);
+ // Add stack trace if the error is an assertion or debug-level logging is enabled
+ if (type == &AssertError || logAny(logLevelDebug))
+ {
+ CHECK(stack != NULL);
- strCat(message, LF_STR);
- strCat(message, stack);
- }
+ strCat(message, LF_STR);
+ strCat(message, stack);
+ }
- THROWP(type, strZ(message));
+ THROWP(type, strZ(message));
+ }
}
+ MEM_CONTEXT_TEMP_END();
FUNCTION_LOG_RETURN_VOID();
}
diff --git a/src/protocol/server.c b/src/protocol/server.c
index 91a698a1e..1319d1ddb 100644
--- a/src/protocol/server.c
+++ b/src/protocol/server.c
@@ -90,15 +90,19 @@ protocolServerError(ProtocolServer *this, int code, const String *message, const
ASSERT(message != NULL);
ASSERT(stack != NULL);
- // Write the error and flush to be sure it gets sent immediately
- PackWrite *error = pckWriteNew(this->write);
- pckWriteU32P(error, protocolMessageTypeError);
- pckWriteI32P(error, code);
- pckWriteStrP(error, message);
- pckWriteStrP(error, stack);
- pckWriteEndP(error);
+ MEM_CONTEXT_TEMP_BEGIN()
+ {
+ // Write the error and flush to be sure it gets sent immediately
+ PackWrite *error = pckWriteNew(this->write);
+ pckWriteU32P(error, protocolMessageTypeError);
+ pckWriteI32P(error, code);
+ pckWriteStrP(error, message);
+ pckWriteStrP(error, stack);
+ pckWriteEndP(error);
- ioWriteFlush(this->write);
+ ioWriteFlush(this->write);
+ }
+ MEM_CONTEXT_TEMP_END();
FUNCTION_LOG_RETURN_VOID();
}
@@ -306,19 +310,23 @@ protocolServerDataPut(ProtocolServer *const this, PackWrite *const data)
FUNCTION_LOG_PARAM(PACK_WRITE, data);
FUNCTION_LOG_END();
- // End the pack
- if (data != NULL)
- pckWriteEndP(data);
-
- // Write the result
- PackWrite *resultMessage = pckWriteNew(this->write);
- pckWriteU32P(resultMessage, protocolMessageTypeData, .defaultWrite = true);
- pckWritePackP(resultMessage, data);
- pckWriteEndP(resultMessage);
-
- // Flush on NULL result since it might be used to synchronize
- if (data == NULL)
- ioWriteFlush(this->write);
+ MEM_CONTEXT_TEMP_BEGIN()
+ {
+ // End the pack
+ if (data != NULL)
+ pckWriteEndP(data);
+
+ // Write the result
+ PackWrite *resultMessage = pckWriteNew(this->write);
+ pckWriteU32P(resultMessage, protocolMessageTypeData, .defaultWrite = true);
+ pckWritePackP(resultMessage, data);
+ pckWriteEndP(resultMessage);
+
+ // Flush on NULL result since it might be used to synchronize
+ if (data == NULL)
+ ioWriteFlush(this->write);
+ }
+ MEM_CONTEXT_TEMP_END();
FUNCTION_LOG_RETURN_VOID();
}
@@ -331,10 +339,14 @@ protocolServerDataEndPut(ProtocolServer *const this)
FUNCTION_LOG_PARAM(PROTOCOL_SERVER, this);
FUNCTION_LOG_END();
- // Write the response and flush to be sure it gets sent immediately
- PackWrite *response = pckWriteNew(this->write);
- pckWriteU32P(response, protocolMessageTypeDataEnd, .defaultWrite = true);
- pckWriteEndP(response);
+ MEM_CONTEXT_TEMP_BEGIN()
+ {
+ // Write the response and flush to be sure it gets sent immediately
+ PackWrite *response = pckWriteNew(this->write);
+ pckWriteU32P(response, protocolMessageTypeDataEnd, .defaultWrite = true);
+ pckWriteEndP(response);
+ }
+ MEM_CONTEXT_TEMP_END();
ioWriteFlush(this->write);
diff --git a/src/storage/remote/protocol.c b/src/storage/remote/protocol.c
index 037e1b97f..4831f8806 100644
--- a/src/storage/remote/protocol.c
+++ b/src/storage/remote/protocol.c
@@ -365,9 +365,13 @@ storageRemoteOpenReadProtocol(PackRead *const param, ProtocolServer *const serve
if (!bufEmpty(buffer))
{
- PackWrite *write = protocolPackNew();
- pckWriteBinP(write, buffer);
- protocolServerDataPut(server, write);
+ MEM_CONTEXT_TEMP_BEGIN()
+ {
+ PackWrite *write = pckWriteNewBuf(bufNew(ioBufferSize() + PROTOCOL_PACK_DEFAULT_SIZE));
+ pckWriteBinP(write, buffer);
+ protocolServerDataPut(server, write);
+ }
+ MEM_CONTEXT_TEMP_END();
bufUsedZero(buffer);
}
diff --git a/src/storage/remote/write.c b/src/storage/remote/write.c
index a7d5b4e41..587e233ee 100644
--- a/src/storage/remote/write.c
+++ b/src/storage/remote/write.c
@@ -5,6 +5,7 @@ Remote Storage File write
#include "common/compress/helper.h"
#include "common/debug.h"
+#include "common/io/io.h"
#include "common/io/write.h"
#include "common/log.h"
#include "common/memContext.h"
@@ -132,7 +133,12 @@ storageWriteRemote(THIS_VOID, const Buffer *buffer)
ASSERT(this != NULL);
ASSERT(buffer != NULL);
- protocolClientDataPut(this->client, pckWriteBinP(protocolPackNew(), buffer));
+ MEM_CONTEXT_TEMP_BEGIN()
+ {
+ protocolClientDataPut(
+ this->client, pckWriteBinP(pckWriteNewBuf(bufNew(ioBufferSize() + PROTOCOL_PACK_DEFAULT_SIZE)), buffer));
+ }
+ MEM_CONTEXT_TEMP_END();
#ifdef DEBUG
this->protocolWriteBytes += bufUsed(buffer);
@@ -158,10 +164,14 @@ storageWriteRemoteClose(THIS_VOID)
// Close if the file has not already been closed
if (this->client != NULL)
{
- protocolClientDataPut(this->client, NULL);
- ioFilterGroupResultAllSet(
- ioWriteFilterGroup(storageWriteIo(this->write)), jsonToVar(pckReadStrP(protocolClientDataGet(this->client))));
- protocolClientDataEndGet(this->client);
+ MEM_CONTEXT_TEMP_BEGIN()
+ {
+ protocolClientDataPut(this->client, NULL);
+ ioFilterGroupResultAllSet(
+ ioWriteFilterGroup(storageWriteIo(this->write)), jsonToVar(pckReadStrP(protocolClientDataGet(this->client))));
+ protocolClientDataEndGet(this->client);
+ }
+ MEM_CONTEXT_TEMP_END();
this->client = NULL;
memContextCallbackClear(this->memContext);
[parent: 33775e53c916]