You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@celeborn.apache.org by "TheodoreLx (via GitHub)" <gi...@apache.org> on 2025/07/28 08:10:33 UTC

[PR] [CELEBORN-2085] Use a fixed buffer for flush copying to reduce GC [celeborn]

TheodoreLx opened a new pull request, #3394:
URL: https://github.com/apache/celeborn/pull/3394

   ### What changes were proposed in this pull request?
   
   Apply for a byte array in advance and use it as a transfer when copying is needed during flush
   
   ### Why are the changes needed?
   
   For HdfsFlushTask, OssFlushTask, and S3FlushTask, you need to copy the CompositeByteBuf in the parameter to a byte array when flushing, and then use the respective clients to write the byte array to the storage.
   When the flush throughput rate is very high, this copying will cause very serious GC problems and affect the performance of the worker
   
   
   ### Does this PR introduce _any_ user-facing change?
   
   no
   
   ### How was this patch tested?
   
   cluster test


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [CELEBORN-2085] Use a fixed buffer for flush copying to reduce GC [celeborn]

Posted by "RexXiong (via GitHub)" <gi...@apache.org>.
RexXiong closed pull request #3394: [CELEBORN-2085] Use a fixed buffer for flush copying to reduce GC
URL: https://github.com/apache/celeborn/pull/3394


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [CELEBORN-2085] Use a fixed buffer for flush copying to reduce GC [celeborn]

Posted by "FMX (via GitHub)" <gi...@apache.org>.
FMX commented on PR #3394:
URL: https://github.com/apache/celeborn/pull/3394#issuecomment-3134626026

   There are two points to be modified,
   1. Flush tasks will most likely be greater than the size of `celeborn.worker.flusher.buffer.size`.
   2. Local flushers don't need this optimization because we don't want to copy memory from off-heap to heap.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [CELEBORN-2085] Use a fixed buffer for flush copying to reduce GC [celeborn]

Posted by "RexXiong (via GitHub)" <gi...@apache.org>.
RexXiong commented on code in PR #3394:
URL: https://github.com/apache/celeborn/pull/3394#discussion_r2246851625


##########
common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala:
##########
@@ -6613,4 +6615,16 @@ object CelebornConf extends Logging {
       .version("0.6.0")
       .booleanConf
       .createWithDefaultString("false")
+
+  val WORKER_FLUSH_REUSE_COPY_BUFFER_ENABLED: ConfigEntry[Boolean] =
+    buildConf("worker.flush.reuse.copy.buffer.enabled")

Review Comment:
   `worker.flush.reuse.copy.buffer.enabled` -> `celeborn.worker.flush.reuseCopyBuffer.enabled`



##########
common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala:
##########
@@ -6613,4 +6615,16 @@ object CelebornConf extends Logging {
       .version("0.6.0")
       .booleanConf
       .createWithDefaultString("false")
+
+  val WORKER_FLUSH_REUSE_COPY_BUFFER_ENABLED: ConfigEntry[Boolean] =
+    buildConf("worker.flush.reuse.copy.buffer.enabled")
+      .categories("worker")
+      .doc("Whether to enable reuse copy buffer for flush. Note that this copy buffer must not" +
+        " be referenced again after flushing. This means that, for example, the Hdfs(Oss or S3) client" +
+        " will not asynchronously access this buffer after the flush method returns, otherwise data" +
+        " modification problems will occur.")
+      .version("0.6.0")

Review Comment:
   0.6.1?



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [CELEBORN-2085] Use a fixed buffer for flush copying to reduce GC [celeborn]

Posted by "TheodoreLx (via GitHub)" <gi...@apache.org>.
TheodoreLx commented on PR #3394:
URL: https://github.com/apache/celeborn/pull/3394#issuecomment-3135050583

   1. If the buffer length in FlushTask is greater than copyBytes, perhaps we can temporarily fall back to the original ByteBufUtil.getBytes
   2. For Local Flusher, we do not allocate byte array to copyBytes in the init method.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [CELEBORN-2085] Use a fixed buffer for flush copying to reduce GC [celeborn]

Posted by "TheodoreLx (via GitHub)" <gi...@apache.org>.
TheodoreLx commented on code in PR #3394:
URL: https://github.com/apache/celeborn/pull/3394#discussion_r2241671317


##########
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/FlushTask.scala:
##########
@@ -33,7 +33,19 @@ abstract private[worker] class FlushTask(
     val notifier: FlushNotifier,
     val keepBuffer: Boolean,
     val source: AbstractSource) {
-  def flush(): Unit
+  def flush(copyBytes: Array[Byte]): Unit
+
+  def convertBufferToBytes(
+      buffer: CompositeByteBuf,
+      copyBytes: Array[Byte],
+      length: Int): Array[Byte] = {
+    if (copyBytes != null) {
+      buffer.readBytes(copyBytes, 0, length)

Review Comment:
   Taking HdfsFlushTask as an example, the capacity of copyBytes is equal to `celeborn.worker.flusher.hdfs.buffer.size`. Generally, the length of the buffer in HdfsFlushTask is less than `celeborn.worker.flusher.hdfs.buffer.size`, so the copy will not be truncated or out of bounds.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [CELEBORN-2085] Use a fixed buffer for flush copying to reduce GC [celeborn]

Posted by "RexXiong (via GitHub)" <gi...@apache.org>.
RexXiong commented on PR #3394:
URL: https://github.com/apache/celeborn/pull/3394#issuecomment-3166667372

   Thanks, merge to main(v0.7.0) and branch-0.6(0.6.1)


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [CELEBORN-2085] Use a fixed buffer for flush copying to reduce GC [celeborn]

Posted by "FMX (via GitHub)" <gi...@apache.org>.
FMX commented on code in PR #3394:
URL: https://github.com/apache/celeborn/pull/3394#discussion_r2244733368


##########
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/Flusher.scala:
##########
@@ -141,13 +147,17 @@ private[worker] class LocalFlusher(
     maxComponents: Int,
     val mountPoint: String,
     val diskType: StorageInfo.Type,
-    timeWindow: TimeWindow) extends Flusher(
+    timeWindow: TimeWindow,
+    reuseCopyBuffer: Boolean,

Review Comment:
   You should set `reuseCopyBuffer` to false if you don't want the local flusher to get involved with the unnecessary memory copy operations.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [CELEBORN-2085] Use a fixed buffer for flush copying to reduce GC [celeborn]

Posted by "FMX (via GitHub)" <gi...@apache.org>.
FMX commented on code in PR #3394:
URL: https://github.com/apache/celeborn/pull/3394#discussion_r2244737033


##########
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/StorageManager.scala:
##########
@@ -157,7 +157,9 @@ final private[worker] class StorageManager(conf: CelebornConf, workerSource: Abs
           conf.workerPushMaxComponents,
           diskInfo.mountPoint,
           diskInfo.storageType,
-          diskInfo.flushTimeMetrics)
+          diskInfo.flushTimeMetrics,
+          conf.workerFlushReuseCopyBufferEnabled,

Review Comment:
   You passed the config parameter to the local flusher, which is the opposite of what you described.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [CELEBORN-2085] Use a fixed buffer for flush copying to reduce GC [celeborn]

Posted by "RexXiong (via GitHub)" <gi...@apache.org>.
RexXiong commented on code in PR #3394:
URL: https://github.com/apache/celeborn/pull/3394#discussion_r2238650417


##########
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/FlushTask.scala:
##########
@@ -33,7 +33,19 @@ abstract private[worker] class FlushTask(
     val notifier: FlushNotifier,
     val keepBuffer: Boolean,
     val source: AbstractSource) {
-  def flush(): Unit
+  def flush(copyBytes: Array[Byte]): Unit
+
+  def convertBufferToBytes(
+      buffer: CompositeByteBuf,
+      copyBytes: Array[Byte],
+      length: Int): Array[Byte] = {
+    if (copyBytes != null) {
+      buffer.readBytes(copyBytes, 0, length)

Review Comment:
   If the length of copyBytes is less than the length of buffer, what will happen?



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org