You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@celeborn.apache.org by "FMX (via GitHub)" <gi...@apache.org> on 2023/03/02 06:46:08 UTC
[GitHub] [incubator-celeborn] FMX commented on a diff in pull request #1296: [CELEBORN-348] Support fetchTime in load-aware slots assignment strategy
FMX commented on code in PR #1296:
URL: https://github.com/apache/incubator-celeborn/pull/1296#discussion_r1122620614
##########
common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala:
##########
@@ -718,9 +722,11 @@ class CelebornConf(loadDefaults: Boolean) extends Cloneable with Logging with Se
def hddFlusherThreads: Int = get(WORKER_FLUSHER_HDD_THREADS)
def ssdFlusherThreads: Int = get(WORKER_FLUSHER_SSD_THREADS)
def hdfsFlusherThreads: Int = get(WORKER_FLUSHER_HDFS_THREADS)
- def avgFlushTimeSlidingWindowSize: Int = get(WORKER_FLUSHER_AVGFLUSHTIME_SLIDINGWINDOW_SIZE)
- def avgFlushTimeSlidingWindowMinCount: Int =
- get(WORKER_FLUSHER_AVGFLUSHTIME_SLIDINGWINDOW_MINCOUNT)
+ def diskTimeSlidingWindowSize: Int = get(WORKER_DISK_TIME_SLIDINGWINDOW_SIZE)
Review Comment:
Unify config style.
##########
common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala:
##########
@@ -1988,6 +2006,24 @@ object CelebornConf extends Logging {
.doubleConf
.createWithDefault(0.1)
+ val SLOTS_ASSIGN_LOADAWARE_FLUSHTIME_WEIGHT: ConfigEntry[Double] =
+ buildConf("celeborn.slots.assign.loadAware.flushTimeWeight")
+ .categories("master")
+ .doc(
+ "Weight of average flush time when calculating ordering in load-aware assignment strategy")
+ .version("0.2.1")
+ .doubleConf
+ .createWithDefault(0)
+
+ val SLOTS_ASSIGN_LOADAWARE_FETCHTIME_WEIGHT: ConfigEntry[Double] =
+ buildConf("celeborn.slots.assign.loadAware.fetchTimeWeight")
+ .categories("master")
+ .doc(
+ "Weight of average fetch time when calculating ordering in load-aware assignment strategy")
+ .version("0.2.1")
+ .doubleConf
Review Comment:
ditto
##########
common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala:
##########
@@ -1948,25 +1954,37 @@ object CelebornConf extends Logging {
.bytesConf(ByteUnit.BYTE)
.createWithDefaultString("5G")
- val WORKER_FLUSHER_AVGFLUSHTIME_SLIDINGWINDOW_SIZE: ConfigEntry[Int] =
- buildConf("celeborn.worker.flusher.avgFlushTime.slidingWindow.size")
+ val WORKER_DISK_TIME_SLIDINGWINDOW_SIZE: ConfigEntry[Int] =
+ buildConf("celeborn.worker.diskTime.slidingWindow.size")
+ .withAlternative("celeborn.worker.flusher.avgFlushTime.slidingWindow.size")
.withAlternative("rss.flusher.avg.time.window")
.categories("worker")
.doc("The size of sliding windows used to calculate statistics about flushed time and count.")
.version("0.2.0")
.intConf
.createWithDefault(20)
- val WORKER_FLUSHER_AVGFLUSHTIME_SLIDINGWINDOW_MINCOUNT: ConfigEntry[Int] =
- buildConf("celeborn.worker.flusher.avgFlushTime.slidingWindow.minCount")
+ val WORKER_DISKTIME_SLIDINGWINDOW_MINFLUSHCOUNT: ConfigEntry[Int] =
+ buildConf("celeborn.worker.diskTime.slidingWindow.minFlushCount")
+ .withAlternative("celeborn.worker.flusher.avgFlushTime.slidingWindow.minCount")
.withAlternative("rss.flusher.avg.time.minimum.count")
.categories("worker")
.doc("The minimum flush count to enter a sliding window" +
" to calculate statistics about flushed time and count.")
.version("0.2.0")
Review Comment:
ditto
##########
common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala:
##########
@@ -1948,25 +1954,37 @@ object CelebornConf extends Logging {
.bytesConf(ByteUnit.BYTE)
.createWithDefaultString("5G")
- val WORKER_FLUSHER_AVGFLUSHTIME_SLIDINGWINDOW_SIZE: ConfigEntry[Int] =
- buildConf("celeborn.worker.flusher.avgFlushTime.slidingWindow.size")
+ val WORKER_DISK_TIME_SLIDINGWINDOW_SIZE: ConfigEntry[Int] =
+ buildConf("celeborn.worker.diskTime.slidingWindow.size")
+ .withAlternative("celeborn.worker.flusher.avgFlushTime.slidingWindow.size")
.withAlternative("rss.flusher.avg.time.window")
.categories("worker")
.doc("The size of sliding windows used to calculate statistics about flushed time and count.")
.version("0.2.0")
Review Comment:
Config name is changed, the version should be changed to 0.3.0 or 0.2.1.
##########
common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala:
##########
@@ -1948,25 +1954,37 @@ object CelebornConf extends Logging {
.bytesConf(ByteUnit.BYTE)
.createWithDefaultString("5G")
- val WORKER_FLUSHER_AVGFLUSHTIME_SLIDINGWINDOW_SIZE: ConfigEntry[Int] =
- buildConf("celeborn.worker.flusher.avgFlushTime.slidingWindow.size")
+ val WORKER_DISK_TIME_SLIDINGWINDOW_SIZE: ConfigEntry[Int] =
+ buildConf("celeborn.worker.diskTime.slidingWindow.size")
+ .withAlternative("celeborn.worker.flusher.avgFlushTime.slidingWindow.size")
.withAlternative("rss.flusher.avg.time.window")
.categories("worker")
.doc("The size of sliding windows used to calculate statistics about flushed time and count.")
.version("0.2.0")
.intConf
.createWithDefault(20)
- val WORKER_FLUSHER_AVGFLUSHTIME_SLIDINGWINDOW_MINCOUNT: ConfigEntry[Int] =
- buildConf("celeborn.worker.flusher.avgFlushTime.slidingWindow.minCount")
+ val WORKER_DISKTIME_SLIDINGWINDOW_MINFLUSHCOUNT: ConfigEntry[Int] =
+ buildConf("celeborn.worker.diskTime.slidingWindow.minFlushCount")
+ .withAlternative("celeborn.worker.flusher.avgFlushTime.slidingWindow.minCount")
.withAlternative("rss.flusher.avg.time.minimum.count")
.categories("worker")
.doc("The minimum flush count to enter a sliding window" +
" to calculate statistics about flushed time and count.")
.version("0.2.0")
.internal
.intConf
- .createWithDefault(1000)
+ .createWithDefault(500)
+
+ val WORKER_DISKTIME_SLIDINGWINDOW_MINFETCHCOUNT: ConfigEntry[Int] =
+ buildConf("celeborn.worker.diskTime.slidingWindow.minFetchCount")
+ .categories("worker")
+ .doc("The minimum fetch count to enter a sliding window" +
+ " to calculate statistics about flushed time and count.")
+ .version("0.2.0")
Review Comment:
ditto.
##########
common/src/main/scala/org/apache/celeborn/common/meta/TimeWindow.scala:
##########
@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.common.meta
+
+import java.util.concurrent.atomic.LongAdder
+
+class TimeWindow(windowSize: Int, minWindowCount: Int) {
+ val totalCount = new LongAdder
+ val totalTime = new LongAdder
+ val timeWindow = new Array[(Long, Long)](windowSize)
+ var index = 0
+
+ for (i <- 0 until windowSize) {
+ timeWindow(i) = (0L, 0L)
+ }
+
+ def update(delta: Long): Unit = {
+ totalTime.add(delta)
+ totalCount.increment()
+ }
+
+ def getAverage(): Long = {
+ val currentFlushTime = totalTime.sumThenReset()
Review Comment:
This variable might be totalFetchTime, so I think the variable name can be changed.
##########
common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala:
##########
@@ -1988,6 +2006,24 @@ object CelebornConf extends Logging {
.doubleConf
.createWithDefault(0.1)
+ val SLOTS_ASSIGN_LOADAWARE_FLUSHTIME_WEIGHT: ConfigEntry[Double] =
+ buildConf("celeborn.slots.assign.loadAware.flushTimeWeight")
+ .categories("master")
+ .doc(
+ "Weight of average flush time when calculating ordering in load-aware assignment strategy")
+ .version("0.2.1")
+ .doubleConf
+ .createWithDefault(0)
Review Comment:
Here can be a range validation between 0 and 1.
##########
common/src/main/scala/org/apache/celeborn/common/meta/TimeWindow.scala:
##########
@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.common.meta
+
+import java.util.concurrent.atomic.LongAdder
+
+class TimeWindow(windowSize: Int, minWindowCount: Int) {
Review Comment:
1. I think this class can be renamed to SlidingWindow.
2. I think minWindowCount is not readable because it is a threshold to put the latest metric into a window. So I think renaming it to threshold will be clear.
##########
master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala:
##########
@@ -510,7 +510,9 @@ private[celeborn] class Master(
requestSlots.shouldReplicate,
diskReserveSize,
slotsAssignLoadAwareDiskGroupNum,
- slotsAssignLoadAwareDiskGroupGradient)
+ slotsAssignLoadAwareDiskGroupGradient,
+ conf.slotsAssignLoadAwareFlushTimeWeight,
Review Comment:
According to our previous code, I think adding a variable or changing the existing code style will be better.
--
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