This is an automated email from the ASF dual-hosted git repository.
adoroszlai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ozone.git
The following commit(s) were added to refs/heads/master by this push:
new a9504b0ebb HDDS-10003. Cache serialized block token in output stream
to reduce heap consumption (#5908)
a9504b0ebb is described below
commit a9504b0ebb93da9e6648e1f78a912749a96ca015
Author: Wei-Chiu Chuang <[email protected]>
AuthorDate: Fri Jan 12 01:12:45 2024 -0800
HDDS-10003. Cache serialized block token in output stream to reduce heap
consumption (#5908)
---
.../hdds/scm/storage/BlockDataStreamOutput.java | 11 +++--
.../hadoop/hdds/scm/storage/BlockOutputStream.java | 11 +++--
.../hdds/scm/storage/ECBlockOutputStream.java | 2 +-
.../hdds/scm/storage/ContainerProtocolCalls.java | 56 +++++-----------------
4 files changed, 28 insertions(+), 52 deletions(-)
diff --git
a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockDataStreamOutput.java
b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockDataStreamOutput.java
index 58bea6a99b..f878659afa 100644
---
a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockDataStreamOutput.java
+++
b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockDataStreamOutput.java
@@ -131,6 +131,7 @@ public class BlockDataStreamOutput implements
ByteBufferStreamOutput {
//number of buffers used before doing a flush/putBlock.
private int flushPeriod;
private final Token<? extends TokenIdentifier> token;
+ private final String tokenString;
private final DataStreamOutput out;
private CompletableFuture<DataStreamReply> dataStreamCloseReply;
private List<CompletableFuture<DataStreamReply>> futures = new ArrayList<>();
@@ -168,6 +169,8 @@ public class BlockDataStreamOutput implements
ByteBufferStreamOutput {
this.xceiverClient =
(XceiverClientRatis)xceiverClientManager.acquireClient(pipeline, true);
this.token = token;
+ this.tokenString = (this.token == null) ? null :
+ this.token.encodeToUrlString();
// Alternatively, stream setup can be delayed till the first chunk write.
this.out = setupStream(pipeline);
this.bufferList = bufferList;
@@ -207,8 +210,8 @@ public class BlockDataStreamOutput implements
ByteBufferStreamOutput {
.setContainerID(blockID.get().getContainerID())
.setDatanodeUuid(id).setWriteChunk(writeChunkRequest);
- if (token != null) {
- builder.setEncodedToken(token.encodeToUrlString());
+ if (tokenString != null) {
+ builder.setEncodedToken(tokenString);
}
ContainerCommandRequestMessage message =
@@ -420,7 +423,7 @@ public class BlockDataStreamOutput implements
ByteBufferStreamOutput {
if (close) {
final ContainerCommandRequestProto putBlockRequest
= ContainerProtocolCalls.getPutBlockRequest(
- xceiverClient.getPipeline(), blockData, true, token);
+ xceiverClient.getPipeline(), blockData, true, tokenString);
dataStreamCloseReply = executePutBlockClose(putBlockRequest,
PUT_BLOCK_REQUEST_LENGTH_MAX, out);
dataStreamCloseReply.whenComplete((reply, e) -> {
@@ -437,7 +440,7 @@ public class BlockDataStreamOutput implements
ByteBufferStreamOutput {
try {
XceiverClientReply asyncReply =
- putBlockAsync(xceiverClient, blockData, close, token);
+ putBlockAsync(xceiverClient, blockData, close, tokenString);
final CompletableFuture<ContainerCommandResponseProto> flushFuture
= asyncReply.getResponse().thenApplyAsync(e -> {
try {
diff --git
a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockOutputStream.java
b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockOutputStream.java
index b97165084f..ac21411ea5 100644
---
a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockOutputStream.java
+++
b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockOutputStream.java
@@ -124,6 +124,7 @@ public class BlockOutputStream extends OutputStream {
//current buffer allocated to write
private ChunkBuffer currentBuffer;
private final Token<? extends TokenIdentifier> token;
+ private final String tokenString;
private int replicationIndex;
private Pipeline pipeline;
private final ContainerClientMetrics clientMetrics;
@@ -166,6 +167,8 @@ public class BlockOutputStream extends OutputStream {
this.xceiverClient = xceiverClientManager.acquireClient(pipeline);
this.bufferPool = bufferPool;
this.token = token;
+ this.tokenString = (this.token == null) ? null :
+ this.token.encodeToUrlString();
//number of buffers used before doing a flush
refreshCurrentBuffer();
@@ -244,8 +247,8 @@ public class BlockOutputStream extends OutputStream {
return this.pipeline;
}
- Token<? extends TokenIdentifier> getToken() {
- return this.token;
+ protected String getTokenString() {
+ return this.tokenString;
}
ExecutorService getResponseExecutor() {
@@ -466,7 +469,7 @@ public class BlockOutputStream extends OutputStream {
try {
BlockData blockData = containerBlockData.build();
XceiverClientReply asyncReply =
- putBlockAsync(xceiverClient, blockData, close, token);
+ putBlockAsync(xceiverClient, blockData, close, tokenString);
CompletableFuture<ContainerProtos.ContainerCommandResponseProto> future =
asyncReply.getResponse();
flushFuture = future.thenApplyAsync(e -> {
@@ -724,7 +727,7 @@ public class BlockOutputStream extends OutputStream {
try {
XceiverClientReply asyncReply = writeChunkAsync(xceiverClient, chunkInfo,
- blockID.get(), data, token, replicationIndex);
+ blockID.get(), data, tokenString, replicationIndex);
CompletableFuture<ContainerProtos.ContainerCommandResponseProto>
respFuture = asyncReply.getResponse();
CompletableFuture<ContainerProtos.ContainerCommandResponseProto>
diff --git
a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ECBlockOutputStream.java
b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ECBlockOutputStream.java
index 1d7fdc1df6..0abc2274bf 100644
---
a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ECBlockOutputStream.java
+++
b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ECBlockOutputStream.java
@@ -228,7 +228,7 @@ public class ECBlockOutputStream extends BlockOutputStream {
try {
ContainerProtos.BlockData blockData = getContainerBlockData().build();
XceiverClientReply asyncReply =
- putBlockAsync(getXceiverClient(), blockData, close, getToken());
+ putBlockAsync(getXceiverClient(), blockData, close,
getTokenString());
CompletableFuture<ContainerProtos.ContainerCommandResponseProto> future =
asyncReply.getResponse();
flushFuture = future.thenApplyAsync(e -> {
diff --git
a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/storage/ContainerProtocolCalls.java
b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/storage/ContainerProtocolCalls.java
index 638282b30c..d20b5e8f7a 100644
---
a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/storage/ContainerProtocolCalls.java
+++
b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/storage/ContainerProtocolCalls.java
@@ -243,58 +243,29 @@ public final class ContainerProtocolCalls {
return response.getGetCommittedBlockLength();
}
- /**
- * Calls the container protocol to put a container block.
- *
- * @param xceiverClient client to perform call
- * @param containerBlockData block data to identify container
- * @param token a token for this block (may be null)
- * @return putBlockResponse
- * @throws IOException if there is an I/O error while performing the call
- */
- public static ContainerProtos.PutBlockResponseProto putBlock(
- XceiverClientSpi xceiverClient, BlockData containerBlockData,
- Token<OzoneBlockTokenIdentifier> token)
- throws IOException {
- PutBlockRequestProto.Builder createBlockRequest =
- PutBlockRequestProto.newBuilder().setBlockData(containerBlockData);
- String id = xceiverClient.getPipeline().getFirstNode().getUuidString();
- ContainerCommandRequestProto.Builder builder =
- ContainerCommandRequestProto.newBuilder().setCmdType(Type.PutBlock)
- .setContainerID(containerBlockData.getBlockID().getContainerID())
- .setDatanodeUuid(id)
- .setPutBlock(createBlockRequest);
- if (token != null) {
- builder.setEncodedToken(token.encodeToUrlString());
- }
- ContainerCommandRequestProto request = builder.build();
- ContainerCommandResponseProto response =
- xceiverClient.sendCommand(request, getValidatorList());
- return response.getPutBlock();
- }
-
/**
* Calls the container protocol to put a container block.
*
* @param xceiverClient client to perform call
* @param containerBlockData block data to identify container
* @param eof whether this is the last putBlock for the same block
- * @param token a token for this block (may be null)
+ * @param tokenString a serialized token for this block (may be null)
* @return putBlockResponse
* @throws IOException if there is an error while performing the call
*/
public static XceiverClientReply putBlockAsync(XceiverClientSpi
xceiverClient,
- BlockData containerBlockData, boolean eof,
- Token<? extends TokenIdentifier> token)
+ BlockData containerBlockData,
+ boolean eof,
+ String tokenString)
throws IOException, InterruptedException, ExecutionException {
final ContainerCommandRequestProto request = getPutBlockRequest(
- xceiverClient.getPipeline(), containerBlockData, eof, token);
+ xceiverClient.getPipeline(), containerBlockData, eof, tokenString);
return xceiverClient.sendCommandAsync(request);
}
public static ContainerCommandRequestProto getPutBlockRequest(
Pipeline pipeline, BlockData containerBlockData, boolean eof,
- Token<? extends TokenIdentifier> token) throws IOException {
+ String tokenString) throws IOException {
PutBlockRequestProto.Builder createBlockRequest =
PutBlockRequestProto.newBuilder()
.setBlockData(containerBlockData)
@@ -305,8 +276,8 @@ public final class ContainerProtocolCalls {
.setContainerID(containerBlockData.getBlockID().getContainerID())
.setDatanodeUuid(id)
.setPutBlock(createBlockRequest);
- if (token != null) {
- builder.setEncodedToken(token.encodeToUrlString());
+ if (tokenString != null) {
+ builder.setEncodedToken(tokenString);
}
return builder.build();
}
@@ -387,14 +358,12 @@ public final class ContainerProtocolCalls {
* @param chunk information about chunk to write
* @param blockID ID of the block
* @param data the data of the chunk to write
- * @param token a token for this block (may be null)
+ * @param tokenString serialized block token
* @throws IOException if there is an I/O error while performing the call
*/
public static XceiverClientReply writeChunkAsync(
XceiverClientSpi xceiverClient, ChunkInfo chunk, BlockID blockID,
- ByteString data, Token<? extends TokenIdentifier> token,
- int replicationIndex
- )
+ ByteString data, String tokenString, int replicationIndex)
throws IOException, ExecutionException, InterruptedException {
WriteChunkRequestProto.Builder writeChunkRequest =
WriteChunkRequestProto.newBuilder()
@@ -413,8 +382,9 @@ public final class ContainerProtocolCalls {
.setContainerID(blockID.getContainerID())
.setDatanodeUuid(id)
.setWriteChunk(writeChunkRequest);
- if (token != null) {
- builder.setEncodedToken(token.encodeToUrlString());
+
+ if (tokenString != null) {
+ builder.setEncodedToken(tokenString);
}
ContainerCommandRequestProto request = builder.build();
return xceiverClient.sendCommandAsync(request);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
