You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@celeborn.apache.org by ch...@apache.org on 2023/09/28 11:20:00 UTC
[incubator-celeborn] branch branch-0.3 updated (89779e3b1 -> f4ffcbeee)
This is an automated email from the ASF dual-hosted git repository.
chengpan pushed a change to branch branch-0.3
in repository https://gitbox.apache.org/repos/asf/incubator-celeborn.git
discard 89779e3b1 [CELEBORN-1014] Output log with bound address and port
discard 529e80aa6 [CELEBORN-1013] Shutdown master if initialized failed
discard e6aa55cab [CELEBORN-986] Use formatted log instead of string concat
discard 1fed29a96 [CELEBORN-1008] Adjust push/fetch timeout checker thread pool and tasks
discard 0eeffb093 [CELEBORN-1007] Improve JVM metrics naming and add ThreadStates metrics
discard bf5feb674 [CELEBORN-1009][DOC] CELEBORN_PREFER_JEMALLOC
discard 581923951 [CELEBORN-771][FLINK] Convert PushDataHandShake, RegionFinish, RegionStart to PB
new dd0593343 [CELEBORN-771][FLINK] Convert PushDataHandShake, RegionFinish, RegionStart to PB
new d40b22663 [CELEBORN-1009][DOC] CELEBORN_PREFER_JEMALLOC
new 79198a6ff [CELEBORN-1007] Improve JVM metrics naming and add ThreadStates metrics
new f146f5cb6 [CELEBORN-1008] Adjust push/fetch timeout checker thread pool and tasks
new bc08cc040 [CELEBORN-986] Use formatted log instead of string concat
new fd5ecc1e3 [CELEBORN-1013] Shutdown master if initialized failed
new f4ffcbeee [CELEBORN-1014] Output log with bound address and port
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (89779e3b1)
\
N -- N -- N refs/heads/branch-0.3 (f4ffcbeee)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 7 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../org/apache/celeborn/common/network/protocol/TransportMessage.java | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
[incubator-celeborn] 04/07: [CELEBORN-1008] Adjust push/fetch timeout checker thread pool and tasks
Posted by ch...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
chengpan pushed a commit to branch branch-0.3
in repository https://gitbox.apache.org/repos/asf/incubator-celeborn.git
commit f146f5cb69d8820a3b6caa6dc69caad3c45254d7
Author: onebox-li <ly...@163.com>
AuthorDate: Thu Sep 28 10:43:08 2023 +0800
[CELEBORN-1008] Adjust push/fetch timeout checker thread pool and tasks
### What changes were proposed in this pull request?
Only push/data module needs push-timeout-checker, and data module needs fetch-timeout-checker.
Here make push-timeout-checker not to be created in Master/LifeCycleManager, and fetch-timeout-checker in Worker.
The same goes for related timeout checker schedule tasks.
### Why are the changes needed?
Ditto
### Does this PR introduce _any_ user-facing change?
No
### How was this patch tested?
Cluster test
Closes #1940 from onebox-li/checker-dev.
Authored-by: onebox-li <ly...@163.com>
Signed-off-by: zky.zhoukeyong <zk...@alibaba-inc.com>
(cherry picked from commit ef4fc51d5f2d9d1a707e06cac2c64f3a07082e42)
Signed-off-by: zky.zhoukeyong <zk...@alibaba-inc.com>
---
.../network/client/TransportResponseHandler.java | 76 +++++++++++++++-------
1 file changed, 51 insertions(+), 25 deletions(-)
diff --git a/common/src/main/java/org/apache/celeborn/common/network/client/TransportResponseHandler.java b/common/src/main/java/org/apache/celeborn/common/network/client/TransportResponseHandler.java
index 7282fa913..78120a2c1 100644
--- a/common/src/main/java/org/apache/celeborn/common/network/client/TransportResponseHandler.java
+++ b/common/src/main/java/org/apache/celeborn/common/network/client/TransportResponseHandler.java
@@ -79,31 +79,49 @@ public class TransportResponseHandler extends MessageHandler<ResponseMessage> {
this.timeOfLastRequestNs = new AtomicLong(0);
this.pushTimeoutCheckerInterval = conf.pushDataTimeoutCheckIntervalMs();
this.fetchTimeoutCheckerInterval = conf.fetchDataTimeoutCheckIntervalMs();
+
+ String module = conf.getModuleName();
+ boolean checkPushTimeout = false;
+ boolean checkFetchTimeout = false;
+ if (TransportModuleConstants.DATA_MODULE.equals(module)) {
+ checkPushTimeout = true;
+ checkFetchTimeout = true;
+ } else if (TransportModuleConstants.PUSH_MODULE.equals(module)) {
+ checkPushTimeout = true;
+ }
synchronized (TransportResponseHandler.class) {
- if (pushTimeoutChecker == null) {
- pushTimeoutChecker =
- ThreadUtils.newDaemonThreadPoolScheduledExecutor(
- "push-timeout-checker", conf.pushDataTimeoutCheckerThreads());
- }
- if (fetchTimeoutChecker == null) {
- fetchTimeoutChecker =
- ThreadUtils.newDaemonThreadPoolScheduledExecutor(
- "fetch-timeout-checker", conf.fetchDataTimeoutCheckerThreads());
+ if (checkPushTimeout) {
+ if (pushTimeoutChecker == null) {
+ pushTimeoutChecker =
+ ThreadUtils.newDaemonThreadPoolScheduledExecutor(
+ "push-timeout-checker", conf.pushDataTimeoutCheckerThreads());
+ }
+ if (checkFetchTimeout) {
+ if (fetchTimeoutChecker == null) {
+ fetchTimeoutChecker =
+ ThreadUtils.newDaemonThreadPoolScheduledExecutor(
+ "fetch-timeout-checker", conf.fetchDataTimeoutCheckerThreads());
+ }
+ }
}
}
- pushCheckerScheduleFuture =
- pushTimeoutChecker.scheduleAtFixedRate(
- () -> failExpiredPushRequest(),
- pushTimeoutCheckerInterval,
- pushTimeoutCheckerInterval,
- TimeUnit.MILLISECONDS);
+ if (checkPushTimeout) {
+ pushCheckerScheduleFuture =
+ pushTimeoutChecker.scheduleAtFixedRate(
+ () -> failExpiredPushRequest(),
+ pushTimeoutCheckerInterval,
+ pushTimeoutCheckerInterval,
+ TimeUnit.MILLISECONDS);
+ }
- fetchCheckerScheduleFuture =
- fetchTimeoutChecker.scheduleAtFixedRate(
- () -> failExpiredFetchRequest(),
- fetchTimeoutCheckerInterval,
- fetchTimeoutCheckerInterval,
- TimeUnit.MILLISECONDS);
+ if (checkFetchTimeout) {
+ fetchCheckerScheduleFuture =
+ fetchTimeoutChecker.scheduleAtFixedRate(
+ () -> failExpiredFetchRequest(),
+ fetchTimeoutCheckerInterval,
+ fetchTimeoutCheckerInterval,
+ TimeUnit.MILLISECONDS);
+ }
}
public void failExpiredPushRequest() {
@@ -251,8 +269,12 @@ public class TransportResponseHandler extends MessageHandler<ResponseMessage> {
remoteAddress);
failOutstandingRequests(new IOException("Connection from " + remoteAddress + " closed"));
}
- pushCheckerScheduleFuture.cancel(false);
- fetchCheckerScheduleFuture.cancel(false);
+ if (pushCheckerScheduleFuture != null) {
+ pushCheckerScheduleFuture.cancel(false);
+ }
+ if (fetchCheckerScheduleFuture != null) {
+ fetchCheckerScheduleFuture.cancel(false);
+ }
}
@Override
@@ -265,8 +287,12 @@ public class TransportResponseHandler extends MessageHandler<ResponseMessage> {
remoteAddress);
failOutstandingRequests(cause);
}
- pushCheckerScheduleFuture.cancel(false);
- fetchCheckerScheduleFuture.cancel(false);
+ if (pushCheckerScheduleFuture != null) {
+ pushCheckerScheduleFuture.cancel(false);
+ }
+ if (fetchCheckerScheduleFuture != null) {
+ fetchCheckerScheduleFuture.cancel(false);
+ }
}
@Override
[incubator-celeborn] 03/07: [CELEBORN-1007] Improve JVM metrics naming and add ThreadStates metrics
Posted by ch...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
chengpan pushed a commit to branch branch-0.3
in repository https://gitbox.apache.org/repos/asf/incubator-celeborn.git
commit 79198a6ffa842cb448552fd24f61cc5d895c8881
Author: onebox-li <ly...@163.com>
AuthorDate: Thu Sep 28 10:18:37 2023 +0800
[CELEBORN-1007] Improve JVM metrics naming and add ThreadStates metrics
### What changes were proposed in this pull request?
Since we use codahale metrics to expose JVM metrics, the name without prefix is not clear and it‘s not easy to make a grafana template for these metrics because it adds collector name or pool name in names rather than labels.
So here I add jvm metric prefixes, remove pool info from name and obtain the pool name as labels if needed.
And add ThreadStates metrics additionally.
### Why are the changes needed?
Make jvm metrics easy to understand and get template
### Does this PR introduce _any_ user-facing change?
Yes，jvm metrics naming is changed，expose threads state additionally.
change examples like below:
For GarbageCollectorMetricSet, G1-Old-Generation.time -> jvm.gc.time{name="G1-Old-Generation"}
For MemoryUsageGaugeSet, total.init -> jvm.memory.total.init ; pools.Metaspace.usage -> jvm.memory.pools.usage{name="Metaspace"}
For BufferPoolMetricSet, direct.count -> jvm.direct.count
For ThreadStatesGaugeSet, add jvm.thread.count.
For G1, the jvm metrics exposed now:
metrics_jvm_gc_time_Value{name="G1-Old-Generation",role="Worker"} 0 1695731141588
metrics_jvm_gc_count_Value{name="G1-Young-Generation",role="Worker"} 2 1695731141588
metrics_jvm_gc_time_Value{name="G1-Young-Generation",role="Worker"} 74 1695731141588
metrics_jvm_gc_count_Value{name="G1-Old-Generation",role="Worker"} 0 1695731141588
metrics_jvm_heap_committed_Value{role="Worker"} 2109734912 1695731141588
metrics_jvm_non_heap_used_Value{role="Worker"} 47700056 1695731141588
metrics_jvm_heap_used_Value{role="Worker"} 82801184 1695731141588
metrics_jvm_total_committed_Value{role="Worker"} 2160263168 1695731141588
metrics_jvm_total_init_Value{role="Worker"} 2112290816 1695731141588
metrics_jvm_non_heap_max_Value{role="Worker"} -1 1695731141588
metrics_jvm_heap_usage_Value{role="Worker"} 0.009639326483011246 1695731141588
metrics_jvm_total_used_Value{role="Worker"} 130502480 1695731141589
metrics_jvm_heap_init_Value{role="Worker"} 2109734912 1695731141589
metrics_jvm_non_heap_committed_Value{role="Worker"} 50528256 1695731141589
metrics_jvm_non_heap_init_Value{role="Worker"} 2555904 1695731141589
metrics_jvm_non_heap_usage_Value{role="Worker"} -4.7701296E7 1695731141589
metrics_jvm_heap_max_Value{role="Worker"} 8589934592 1695731141589
metrics_jvm_total_max_Value{role="Worker"} 8589934591 1695731141589
metrics_jvm_memory_pool_used_Value{name="Code-Cache",role="Worker"} 10314368 1695731141588
metrics_jvm_memory_pool_committed_Value{name="Code-Cache",role="Worker"} 10944512 1695731141588
metrics_jvm_memory_pool_init_Value{name="G1-Eden-Space",role="Worker"} 111149056 1695731141588
metrics_jvm_memory_pool_max_Value{name="G1-Old-Gen",role="Worker"} 8589934592 1695731141588
metrics_jvm_memory_pool_used_after_gc_Value{name="G1-Survivor-Space",role="Worker"} 14680064 1695731141588
metrics_jvm_memory_pool_used_Value{name="Compressed-Class-Space",role="Worker"} 4440192 1695731141588
metrics_jvm_memory_pool_usage_Value{name="Metaspace",role="Worker"} 0.9449504192610433 1695731141588
metrics_jvm_memory_pool_max_Value{name="Metaspace",role="Worker"} -1 1695731141588
metrics_jvm_memory_pool_init_Value{name="G1-Survivor-Space",role="Worker"} 0 1695731141588
metrics_jvm_memory_pool_committed_Value{name="G1-Old-Gen",role="Worker"} 1998585856 1695731141588
metrics_jvm_memory_pool_committed_Value{name="G1-Survivor-Space",role="Worker"} 14680064 1695731141588
metrics_jvm_memory_pool_committed_Value{name="G1-Eden-Space",role="Worker"} 96468992 1695731141588
metrics_jvm_memory_pool_max_Value{name="G1-Survivor-Space",role="Worker"} -1 1695731141588
metrics_jvm_memory_pool_usage_Value{name="Compressed-Class-Space",role="Worker"} 0.004135251045227051 1695731141588
metrics_jvm_memory_pool_usage_Value{name="G1-Survivor-Space",role="Worker"} 1.0 1695731141588
metrics_jvm_memory_pool_max_Value{name="Code-Cache",role="Worker"} 251658240 1695731141588
metrics_jvm_memory_pool_init_Value{name="Compressed-Class-Space",role="Worker"} 0 1695731141589
metrics_jvm_memory_pool_usage_Value{name="G1-Eden-Space",role="Worker"} 0.34782608695652173 1695731141589
metrics_jvm_memory_pool_init_Value{name="Metaspace",role="Worker"} 0 1695731141589
metrics_jvm_memory_pool_max_Value{name="G1-Eden-Space",role="Worker"} -1 1695731141589
metrics_jvm_memory_pool_usage_Value{name="Code-Cache",role="Worker"} 0.04098917643229167 1695731141589
metrics_jvm_memory_pool_used_after_gc_Value{name="G1-Eden-Space",role="Worker"} 0 1695731141589
metrics_jvm_memory_pool_init_Value{name="Code-Cache",role="Worker"} 2555904 1695731141589
metrics_jvm_memory_pool_used_Value{name="G1-Survivor-Space",role="Worker"} 14680064 1695731141589
metrics_jvm_memory_pool_committed_Value{name="Compressed-Class-Space",role="Worker"} 4718592 1695731141589
metrics_jvm_memory_pool_used_Value{name="G1-Eden-Space",role="Worker"} 33554432 1695731141589
metrics_jvm_memory_pool_used_Value{name="G1-Old-Gen",role="Worker"} 34566688 1695731141589
metrics_jvm_memory_pool_usage_Value{name="G1-Old-Gen",role="Worker"} 0.004024092108011246 1695731141589
metrics_jvm_memory_pool_used_after_gc_Value{name="G1-Old-Gen",role="Worker"} 0 1695731141589
metrics_jvm_memory_pool_committed_Value{name="Metaspace",role="Worker"} 34865152 1695731141589
metrics_jvm_memory_pool_init_Value{name="G1-Old-Gen",role="Worker"} 1998585856 1695731141589
metrics_jvm_memory_pool_used_Value{name="Metaspace",role="Worker"} 32945840 1695731141589
metrics_jvm_memory_pool_max_Value{name="Compressed-Class-Space",role="Worker"} 1073741824 1695731141589
metrics_jvm_direct_count_Value{role="Worker"} 8 1695731141589
metrics_jvm_direct_capacity_Value{role="Worker"} 1036 1695731141589
metrics_jvm_direct_used_Value{role="Worker"} 1037 1695731141589
metrics_jvm_mapped_used_Value{role="Worker"} 0 1695731141589
metrics_jvm_mapped_capacity_Value{role="Worker"} 0 1695731141589
metrics_jvm_mapped_count_Value{role="Worker"} 0 1695731141589
metrics_jvm_thread_timed_waiting_count_Value{role="Worker"} 23 1695731141589
metrics_jvm_thread_deadlock_count_Value{role="Worker"} 0 1695731141589
metrics_jvm_thread_count_Value{role="Worker"} 78 1695731141589
metrics_jvm_thread_waiting_count_Value{role="Worker"} 45 1695731141589
metrics_jvm_thread_daemon_count_Value{role="Worker"} 75 1695731141589
metrics_jvm_thread_new_count_Value{role="Worker"} 0 1695731141589
metrics_jvm_thread_blocked_count_Value{role="Worker"} 0 1695731141590
metrics_jvm_thread_deadlocks_Value{role="Worker"} [] 1695731141590
metrics_jvm_thread_runnable_count_Value{role="Worker"} 10 1695731141590
metrics_jvm_thread_terminated_count_Value{role="Worker"} 0 1695731141590
### How was this patch tested?
UT and cluster test with g1, PS-Scavenge/PS-MarkSweep and ParNew/CMS
Closes #1939 from onebox-li/improve-jvm-metrics.
Authored-by: onebox-li <ly...@163.com>
Signed-off-by: zky.zhoukeyong <zk...@alibaba-inc.com>
(cherry picked from commit b4dfc09dcf528f61da22bea01e66dc3c450ec0b6)
Signed-off-by: zky.zhoukeyong <zk...@alibaba-inc.com>
---
.../celeborn/common/metrics/source/JVMSource.scala | 82 +++++++++++++++++++---
.../common/metrics/source/JVMSourceSuite.scala | 55 +++++++++++++++
2 files changed, 126 insertions(+), 11 deletions(-)
diff --git a/common/src/main/scala/org/apache/celeborn/common/metrics/source/JVMSource.scala b/common/src/main/scala/org/apache/celeborn/common/metrics/source/JVMSource.scala
index 374024dd2..a87924868 100644
--- a/common/src/main/scala/org/apache/celeborn/common/metrics/source/JVMSource.scala
+++ b/common/src/main/scala/org/apache/celeborn/common/metrics/source/JVMSource.scala
@@ -20,26 +20,86 @@ package org.apache.celeborn.common.metrics.source
import java.lang.management.ManagementFactory
import scala.collection.JavaConverters._
+import scala.collection.mutable
-import com.codahale.metrics.Gauge
-import com.codahale.metrics.jvm.{BufferPoolMetricSet, GarbageCollectorMetricSet, MemoryUsageGaugeSet}
+import com.codahale.metrics.{Gauge, MetricRegistry}
+import com.codahale.metrics.jvm.{BufferPoolMetricSet, GarbageCollectorMetricSet, MemoryUsageGaugeSet, ThreadStatesGaugeSet}
import org.apache.celeborn.common.CelebornConf
class JVMSource(conf: CelebornConf, role: String) extends AbstractSource(conf, role) {
override val sourceName = "JVM"
- // all of metrics of GCMetricSet and BufferPoolMetricSet are Gauge
- Seq(
- new GarbageCollectorMetricSet(),
- new MemoryUsageGaugeSet(),
- new BufferPoolMetricSet(ManagementFactory.getPlatformMBeanServer))
- .map { x =>
- x.getMetrics.asScala.map {
- case (name: String, metric: Gauge[_]) => addGauge(name, metric)
- case (name, metric) => new IllegalArgumentException(s"Unknown metric type: $name: $metric")
+ import JVMSource._
+
+ private val gcNames = ManagementFactory.getGarbageCollectorMXBeans.asScala.map(bean =>
+ WHITESPACE.matcher(bean.getName).replaceAll("-"))
+ private val poolNames = ManagementFactory.getMemoryPoolMXBeans.asScala.map(bean =>
+ WHITESPACE.matcher(bean.getName).replaceAll("-"))
+
+ /**
+ * Add jvm metric prefix, remove pool info from name and obtain the pool name as labels if needed
+ * @param metricName metric name from MetricSet
+ * @param targets keywords need to be replaced
+ * @param prefix prefix for new metric name
+ * @param replacement replacement for pool name
+ * @return new metric without target, labels if exists
+ */
+ def handleJVMMetricName(
+ metricName: String,
+ targets: mutable.Buffer[String],
+ prefix: String,
+ replacement: String): (String, Map[String, String]) = {
+ for (target <- targets) {
+ if (metricName.contains(target)) {
+ val labels = Map("name" -> target)
+ var replaceTarget = target
+ if (replacement.isEmpty) {
+ replaceTarget = target + "."
+ }
+ return (MetricRegistry.name(prefix, metricName.replace(replaceTarget, replacement)), labels)
}
}
+ (MetricRegistry.name(prefix, metricName), Map.empty[String, String])
+ }
+
+ // all metrics in MetricSet are gauges
+ Seq(new GarbageCollectorMetricSet()).map(_.getMetrics.asScala.map {
+ case (name: String, metric: Gauge[_]) =>
+ val newMetrics = handleJVMMetricName(name, gcNames, JVM_METRIC_PREFIX, "gc")
+ addGauge(newMetrics._1, newMetrics._2, metric)
+ case (name, metric) => new IllegalArgumentException(s"Unknown metric type: $name: $metric")
+ })
+
+ Seq(new MemoryUsageGaugeSet()).map(_.getMetrics.asScala.map {
+ case (name: String, metric: Gauge[_]) =>
+ val newMetrics = handleJVMMetricName(name, poolNames, JVM_METRIC_MEMORY_PREFIX, "")
+ addGauge(newMetrics._1, newMetrics._2, metric)
+ case (name, metric) => new IllegalArgumentException(s"Unknown metric type: $name: $metric")
+ })
+
+ Seq(
+ new BufferPoolMetricSet(ManagementFactory.getPlatformMBeanServer)).map(
+ _.getMetrics.asScala.map {
+ case (name: String, metric: Gauge[_]) =>
+ addGauge(MetricRegistry.name(JVM_METRIC_PREFIX, name), metric)
+ case (name, metric) => new IllegalArgumentException(s"Unknown metric type: $name: $metric")
+ })
+
+ Seq(new ThreadStatesGaugeSet()).map(_.getMetrics.asScala.map {
+ case (name: String, metric: Gauge[_]) =>
+ addGauge(MetricRegistry.name(JVM_METRIC_THREAD_PREFIX, name), metric)
+ case (name, metric) => new IllegalArgumentException(s"Unknown metric type: $name: $metric")
+ })
+
// start cleaner
startCleaner()
}
+
+object JVMSource {
+ private val JVM_METRIC_PREFIX = "jvm"
+ private val JVM_METRIC_MEMORY_PREFIX = JVM_METRIC_PREFIX + ".memory"
+ private val JVM_METRIC_THREAD_PREFIX = JVM_METRIC_PREFIX + ".thread"
+
+ private val WHITESPACE = "\\s+".r.pattern
+}
diff --git a/common/src/test/scala/org/apache/celeborn/common/metrics/source/JVMSourceSuite.scala b/common/src/test/scala/org/apache/celeborn/common/metrics/source/JVMSourceSuite.scala
new file mode 100644
index 000000000..6f4f8cac7
--- /dev/null
+++ b/common/src/test/scala/org/apache/celeborn/common/metrics/source/JVMSourceSuite.scala
@@ -0,0 +1,55 @@
+/*
+ * 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.metrics.source
+
+import org.apache.celeborn.CelebornFunSuite
+import org.apache.celeborn.common.CelebornConf
+
+class JVMSourceSuite extends CelebornFunSuite {
+
+ val gcNames = Seq("G1-Young-Generation", "G1-Old-Generation").toBuffer
+ val poolNames = Seq("G1-Eden-Space", "G1-Survivor-Space", "G1-Old-Gen").toBuffer
+
+ val JVM_METRIC_PREFIX = "jvm"
+ val JVM_MEMORY_PREFIX = "jvm.memory"
+
+ test("Test handleJVMMetricName") {
+
+ val jvmSource = new JVMSource(new CelebornConf(), "test")
+
+ val gcMetric1 = "G1-Old-Generation.time"
+ val gcMetric2 = "G1-Young-Generation.count"
+ val gcResult1 = jvmSource.handleJVMMetricName(gcMetric1, gcNames, JVM_METRIC_PREFIX, "gc")
+ val gcResult2 = jvmSource.handleJVMMetricName(gcMetric2, gcNames, JVM_METRIC_PREFIX, "gc")
+ assert(gcResult1._1 == "jvm.gc.time")
+ assert(gcResult1._2 == Map("name" -> "G1-Old-Generation"))
+ assert(gcResult2._1 == "jvm.gc.count")
+ assert(gcResult2._2 == Map("name" -> "G1-Young-Generation"))
+
+ val memoryMetric1 = "total.init"
+ val memoryMetrics = "pools.G1-Eden-Space.init"
+ val memoryResult1 =
+ jvmSource.handleJVMMetricName(memoryMetric1, poolNames, JVM_MEMORY_PREFIX, "")
+ val memoryResult2 =
+ jvmSource.handleJVMMetricName(memoryMetrics, poolNames, JVM_MEMORY_PREFIX, "")
+ assert(memoryResult1._1 == "jvm.memory.total.init")
+ assert(memoryResult1._2 == Map.empty[String, String])
+ assert(memoryResult2._1 == "jvm.memory.pools.init")
+ assert(memoryResult2._2 == Map("name" -> "G1-Eden-Space"))
+ }
+}
[incubator-celeborn] 02/07: [CELEBORN-1009][DOC] CELEBORN_PREFER_JEMALLOC
Posted by ch...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
chengpan pushed a commit to branch branch-0.3
in repository https://gitbox.apache.org/repos/asf/incubator-celeborn.git
commit d40b22663eb233993940997c66d4f73d2f2aba39
Author: sychen <sy...@ctrip.com>
AuthorDate: Wed Sep 27 23:14:27 2023 +0800
[CELEBORN-1009][DOC] CELEBORN_PREFER_JEMALLOC
### What changes were proposed in this pull request?
### Why are the changes needed?
### Does this PR introduce _any_ user-facing change?
### How was this patch tested?
Closes #1944 from cxzl25/CELEBORN-1009.
Lead-authored-by: sychen <sy...@ctrip.com>
Co-authored-by: Keyong Zhou <zh...@apache.org>
Signed-off-by: zky.zhoukeyong <zk...@alibaba-inc.com>
(cherry picked from commit 3e515c5d2e15fdd5d152819f52ca40a674208efc)
Signed-off-by: zky.zhoukeyong <zk...@alibaba-inc.com>
---
docs/configuration/index.md | 28 +++++++++++++++-------------
1 file changed, 15 insertions(+), 13 deletions(-)
diff --git a/docs/configuration/index.md b/docs/configuration/index.md
index 2ee39c40a..fbda2f5ab 100644
--- a/docs/configuration/index.md
+++ b/docs/configuration/index.md
@@ -125,19 +125,21 @@ end="<!--end-include-->"
Recommend configuring in `conf/celeborn-env.sh`.
-| Key | Default | Description |
-|----------------------------------|-------------------------------------------------|-------------|
-| `CELEBORN_HOME` | ``$(cd "`dirname "$0"`"/..; pwd)`` | |
-| `CELEBORN_CONF_DIR` | `${CELEBORN_CONF_DIR:-"${CELEBORN_HOME}/conf"}` | |
-| `CELEBORN_MASTER_MEMORY` | 1 GB | |
-| `CELEBORN_WORKER_MEMORY` | 1 GB | |
-| `CELEBORN_WORKER_OFFHEAP_MEMORY` | 1 GB | |
-| `CELEBORN_MASTER_JAVA_OPTS` | | |
-| `CELEBORN_WORKER_JAVA_OPTS` | | |
-| `CELEBORN_PID_DIR` | `${CELEBORN_HOME}/pids` | |
-| `CELEBORN_LOG_DIR` | `${CELEBORN_HOME}/logs` | |
-| `CELEBORN_SSH_OPTS` | `-o StrictHostKeyChecking=no` | |
-| `CELEBORN_SLEEP` | | |
+| Key | Default | Description |
+|----------------------------------|-------------------------------------------------|--------------------------------------------------------|
+| `CELEBORN_HOME` | ``$(cd "`dirname "$0"`"/..; pwd)`` | |
+| `CELEBORN_CONF_DIR` | `${CELEBORN_CONF_DIR:-"${CELEBORN_HOME}/conf"}` | |
+| `CELEBORN_MASTER_MEMORY` | 1 GB | |
+| `CELEBORN_WORKER_MEMORY` | 1 GB | |
+| `CELEBORN_WORKER_OFFHEAP_MEMORY` | 1 GB | |
+| `CELEBORN_MASTER_JAVA_OPTS` | | |
+| `CELEBORN_WORKER_JAVA_OPTS` | | |
+| `CELEBORN_PID_DIR` | `${CELEBORN_HOME}/pids` | |
+| `CELEBORN_LOG_DIR` | `${CELEBORN_HOME}/logs` | |
+| `CELEBORN_SSH_OPTS` | `-o StrictHostKeyChecking=no` | |
+| `CELEBORN_SLEEP` | | Waiting time for `start-all` and `stop-all` operations |
+| `CELEBORN_PREFER_JEMALLOC` | | set `true` to enable jemalloc memory allocator |
+| `CELEBORN_JEMALLOC_PATH` | | jemalloc library path |
## Tuning
[incubator-celeborn] 05/07: [CELEBORN-986] Use formatted log instead of string concat
Posted by ch...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
chengpan pushed a commit to branch branch-0.3
in repository https://gitbox.apache.org/repos/asf/incubator-celeborn.git
commit bc08cc040db9d3b91bdf2e8cfeec9b8f3ca4f1d1
Author: sychen <sy...@ctrip.com>
AuthorDate: Thu Sep 28 16:29:58 2023 +0800
[CELEBORN-986] Use formatted log instead of string concat
### What changes were proposed in this pull request?
### Why are the changes needed?
### Does this PR introduce _any_ user-facing change?
### How was this patch tested?
GA
Closes #1941 from cxzl25/CELEBORN-986.
Authored-by: sychen <sy...@ctrip.com>
Signed-off-by: zky.zhoukeyong <zk...@alibaba-inc.com>
(cherry picked from commit a9ed7f6a3996148653eaf416900af30de8a83444)
Signed-off-by: zky.zhoukeyong <zk...@alibaba-inc.com>
---
.../celeborn/plugin/flink/RemoteShuffleMaster.java | 2 +-
.../celeborn/plugin/flink/RemoteShuffleMasterTest.java | 2 +-
.../apache/celeborn/common/client/MasterClient.java | 2 +-
.../celeborn/common/util/ShutdownHookManager.java | 18 ++++--------------
.../deploy/master/clustermeta/ha/MetaHandler.java | 4 ++--
5 files changed, 9 insertions(+), 19 deletions(-)
diff --git a/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleMaster.java b/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleMaster.java
index 7bf39d5a2..843c140b0 100644
--- a/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleMaster.java
+++ b/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleMaster.java
@@ -227,7 +227,7 @@ public class RemoteShuffleMaster implements ShuffleMaster<RemoteShuffleDescripto
jobShuffleIds.clear();
lifecycleManager.stop();
} catch (Exception e) {
- LOG.warn("Encounter exception when shutdown: " + e.getMessage(), e);
+ LOG.warn("Encounter exception when shutdown: {}", e.getMessage(), e);
}
ThreadUtils.shutdownExecutors(10, executor);
diff --git a/client-flink/flink-1.14/src/test/java/org/apache/celeborn/plugin/flink/RemoteShuffleMasterTest.java b/client-flink/flink-1.14/src/test/java/org/apache/celeborn/plugin/flink/RemoteShuffleMasterTest.java
index 6768cedb1..a8607a6d8 100644
--- a/client-flink/flink-1.14/src/test/java/org/apache/celeborn/plugin/flink/RemoteShuffleMasterTest.java
+++ b/client-flink/flink-1.14/src/test/java/org/apache/celeborn/plugin/flink/RemoteShuffleMasterTest.java
@@ -93,7 +93,7 @@ public class RemoteShuffleMasterTest {
ShuffleResourceDescriptor mapPartitionShuffleDescriptor =
shuffleResource.getMapPartitionShuffleDescriptor();
- LOG.info("remoteShuffleDescriptor:", remoteShuffleDescriptor);
+ LOG.info("remoteShuffleDescriptor:{}", remoteShuffleDescriptor);
Assert.assertEquals(0, mapPartitionShuffleDescriptor.getShuffleId());
Assert.assertEquals(0, mapPartitionShuffleDescriptor.getPartitionId());
Assert.assertEquals(0, mapPartitionShuffleDescriptor.getAttemptId());
diff --git a/common/src/main/java/org/apache/celeborn/common/client/MasterClient.java b/common/src/main/java/org/apache/celeborn/common/client/MasterClient.java
index 559e409da..92f6061a3 100644
--- a/common/src/main/java/org/apache/celeborn/common/client/MasterClient.java
+++ b/common/src/main/java/org/apache/celeborn/common/client/MasterClient.java
@@ -136,7 +136,7 @@ public class MasterClient {
.requestId_(encodeRequestId(UUID.randomUUID().toString(), nextCallId()));
}
- LOG.debug("Send rpc message " + message);
+ LOG.debug("Send rpc message {}", message);
RpcEndpointRef endpointRef = null;
// Use AtomicInteger or Integer or any Object which holds an int value is ok, we just need to
// transfer an object to get the change of the current index of master addresses.
diff --git a/common/src/main/java/org/apache/celeborn/common/util/ShutdownHookManager.java b/common/src/main/java/org/apache/celeborn/common/util/ShutdownHookManager.java
index d7b794f29..3439a6ba1 100644
--- a/common/src/main/java/org/apache/celeborn/common/util/ShutdownHookManager.java
+++ b/common/src/main/java/org/apache/celeborn/common/util/ShutdownHookManager.java
@@ -115,19 +115,9 @@ public final class ShutdownHookManager {
} catch (TimeoutException ex) {
timeouts++;
future.cancel(true);
- LOG.warn(
- "ShutdownHook '"
- + entry.getHook().getClass().getSimpleName()
- + "' timeout, "
- + ex.toString(),
- ex);
+ LOG.warn("ShutdownHook '{}' timeout", entry.getHook().getClass().getSimpleName(), ex);
} catch (Throwable ex) {
- LOG.warn(
- "ShutdownHook '"
- + entry.getHook().getClass().getSimpleName()
- + "' failed, "
- + ex.toString(),
- ex);
+ LOG.warn("ShutdownHook '{}' failed", entry.getHook().getClass().getSimpleName(), ex);
}
}
return timeouts;
@@ -144,13 +134,13 @@ public final class ShutdownHookManager {
long shutdownTimeout = getShutdownTimeout(conf);
if (!EXECUTOR.awaitTermination(shutdownTimeout, TIME_UNIT_DEFAULT)) {
// timeout waiting for the
- LOG.error("ShutdownHookManger shutdown forcefully after" + " {} seconds.", shutdownTimeout);
+ LOG.error("ShutdownHookManger shutdown forcefully after {} seconds.", shutdownTimeout);
EXECUTOR.shutdownNow();
}
LOG.debug("ShutdownHookManger completed shutdown.");
} catch (InterruptedException ex) {
// interrupted.
- LOG.error("ShutdownHookManger interrupted while waiting for " + "termination.", ex);
+ LOG.error("ShutdownHookManger interrupted while waiting for termination.", ex);
EXECUTOR.shutdownNow();
Thread.currentThread().interrupt();
}
diff --git a/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/MetaHandler.java b/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/MetaHandler.java
index ab9d39dbc..2bbc8ab44 100644
--- a/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/MetaHandler.java
+++ b/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/MetaHandler.java
@@ -74,7 +74,7 @@ public class MetaHandler {
throw new IOException("Can not parse this command!" + request);
}
} catch (IOException e) {
- LOG.warn("Handle meta read request " + cmdType + " failed!", e);
+ LOG.warn("Handle meta read request {} failed!", cmdType, e);
responseBuilder.setSuccess(false);
responseBuilder.setStatus(ResourceProtos.Status.INTERNAL_ERROR);
if (e.getMessage() != null) {
@@ -241,7 +241,7 @@ public class MetaHandler {
}
responseBuilder.setStatus(ResourceProtos.Status.OK);
} catch (IOException e) {
- LOG.warn("Handle meta write request " + cmdType + " failed!", e);
+ LOG.warn("Handle meta write request {} failed!", cmdType, e);
responseBuilder.setSuccess(false);
responseBuilder.setStatus(ResourceProtos.Status.INTERNAL_ERROR);
if (e.getMessage() != null) {
[incubator-celeborn] 01/07: [CELEBORN-771][FLINK] Convert PushDataHandShake, RegionFinish, RegionStart to PB
Posted by ch...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
chengpan pushed a commit to branch branch-0.3
in repository https://gitbox.apache.org/repos/asf/incubator-celeborn.git
commit dd05933430375bd3ed56c3707b7982d567b7f5b8
Author: SteNicholas <pr...@163.com>
AuthorDate: Fri Sep 22 11:36:45 2023 +0800
[CELEBORN-771][FLINK] Convert PushDataHandShake, RegionFinish, RegionStart to PB
### What changes were proposed in this pull request?
`PushDataHandShake`, `RegionFinish`, and `RegionStart` should merge to transport messages to enhance celeborn's compatibility.
### Why are the changes needed?
1. Improves celeborn's transport flexibility to change RPC.
2. Makes Compatible with 0.2 client.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
- `RemoteShuffleOutputGateSuiteJ`
Closes #1910 from SteNicholas/CELEBORN-771.
Authored-by: SteNicholas <pr...@163.com>
Signed-off-by: mingji <fe...@alibaba-inc.com>
(cherry picked from commit 55e85055b80111b6f3662474c8fd500bf6589f2a)
---
.../flink/readclient/FlinkShuffleClientImpl.java | 70 ++++++----
.../common/network/protocol/PushDataHandShake.java | 1 +
.../common/network/protocol/RegionFinish.java | 1 +
.../common/network/protocol/RegionStart.java | 1 +
.../common/network/protocol/TransportMessage.java | 12 ++
common/src/main/proto/TransportMessages.proto | 29 +++++
.../service/deploy/worker/PushDataHandler.scala | 145 +++++++++++++++++----
7 files changed, 207 insertions(+), 52 deletions(-)
diff --git a/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/readclient/FlinkShuffleClientImpl.java b/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/readclient/FlinkShuffleClientImpl.java
index 992716894..c7ed33b33 100644
--- a/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/readclient/FlinkShuffleClientImpl.java
+++ b/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/readclient/FlinkShuffleClientImpl.java
@@ -44,13 +44,16 @@ import org.apache.celeborn.common.network.client.RpcResponseCallback;
import org.apache.celeborn.common.network.client.TransportClient;
import org.apache.celeborn.common.network.client.TransportClientFactory;
import org.apache.celeborn.common.network.protocol.PushData;
-import org.apache.celeborn.common.network.protocol.PushDataHandShake;
-import org.apache.celeborn.common.network.protocol.RegionFinish;
-import org.apache.celeborn.common.network.protocol.RegionStart;
+import org.apache.celeborn.common.network.protocol.TransportMessage;
import org.apache.celeborn.common.network.util.TransportConf;
+import org.apache.celeborn.common.protocol.MessageType;
import org.apache.celeborn.common.protocol.PartitionLocation;
import org.apache.celeborn.common.protocol.PbChangeLocationPartitionInfo;
import org.apache.celeborn.common.protocol.PbChangeLocationResponse;
+import org.apache.celeborn.common.protocol.PbPartitionLocation.Mode;
+import org.apache.celeborn.common.protocol.PbPushDataHandShake;
+import org.apache.celeborn.common.protocol.PbRegionFinish;
+import org.apache.celeborn.common.protocol.PbRegionStart;
import org.apache.celeborn.common.protocol.ReviveRequest;
import org.apache.celeborn.common.protocol.TransportModuleConstants;
import org.apache.celeborn.common.protocol.message.ControlMessages;
@@ -332,18 +335,23 @@ public class FlinkShuffleClientImpl extends ShuffleClientImpl {
location.getUniqueId());
logger.debug("PushDataHandShake location {}", location);
TransportClient client = createClientWaitingInFlightRequest(location, mapKey, pushState);
- PushDataHandShake handShake =
- new PushDataHandShake(
- PRIMARY_MODE,
- shuffleKey,
- location.getUniqueId(),
- attemptId,
- numPartitions,
- bufferSize);
ByteBuffer pushDataHandShakeResponse;
try {
pushDataHandShakeResponse =
- client.sendRpcSync(handShake.toByteBuffer(), conf.pushDataTimeoutMs());
+ client.sendRpcSync(
+ new TransportMessage(
+ MessageType.PUSH_DATA_HAND_SHAKE,
+ PbPushDataHandShake.newBuilder()
+ .setMode(Mode.forNumber(PRIMARY_MODE))
+ .setShuffleKey(shuffleKey)
+ .setPartitionUniqueId(location.getUniqueId())
+ .setAttemptId(attemptId)
+ .setNumPartitions(numPartitions)
+ .setBufferSize(bufferSize)
+ .build()
+ .toByteArray())
+ .toByteBuffer(),
+ conf.pushDataTimeoutMs());
} catch (IOException e) {
// ioexeption revive
return revive(shuffleId, mapId, attemptId, location);
@@ -378,18 +386,23 @@ public class FlinkShuffleClientImpl extends ShuffleClientImpl {
location.getUniqueId());
logger.debug("RegionStart for location {}.", location.toString());
TransportClient client = createClientWaitingInFlightRequest(location, mapKey, pushState);
- RegionStart regionStart =
- new RegionStart(
- PRIMARY_MODE,
- shuffleKey,
- location.getUniqueId(),
- attemptId,
- currentRegionIdx,
- isBroadcast);
ByteBuffer regionStartResponse;
try {
regionStartResponse =
- client.sendRpcSync(regionStart.toByteBuffer(), conf.pushDataTimeoutMs());
+ client.sendRpcSync(
+ new TransportMessage(
+ MessageType.REGION_START,
+ PbRegionStart.newBuilder()
+ .setMode(Mode.forNumber(PRIMARY_MODE))
+ .setShuffleKey(shuffleKey)
+ .setPartitionUniqueId(location.getUniqueId())
+ .setAttemptId(attemptId)
+ .setCurrentRegionIndex(currentRegionIdx)
+ .setIsBroadcast(isBroadcast)
+ .build()
+ .toByteArray())
+ .toByteBuffer(),
+ conf.pushDataTimeoutMs());
} catch (IOException e) {
// ioexeption revive
return revive(shuffleId, mapId, attemptId, location);
@@ -459,9 +472,18 @@ public class FlinkShuffleClientImpl extends ShuffleClientImpl {
location.getUniqueId());
logger.debug("RegionFinish for location {}.", location);
TransportClient client = createClientWaitingInFlightRequest(location, mapKey, pushState);
- RegionFinish regionFinish =
- new RegionFinish(PRIMARY_MODE, shuffleKey, location.getUniqueId(), attemptId);
- client.sendRpcSync(regionFinish.toByteBuffer(), conf.pushDataTimeoutMs());
+ client.sendRpcSync(
+ new TransportMessage(
+ MessageType.REGION_FINISH,
+ PbRegionFinish.newBuilder()
+ .setMode(Mode.forNumber(PRIMARY_MODE))
+ .setShuffleKey(shuffleKey)
+ .setPartitionUniqueId(location.getUniqueId())
+ .setAttemptId(attemptId)
+ .build()
+ .toByteArray())
+ .toByteBuffer(),
+ conf.pushDataTimeoutMs());
return null;
});
}
diff --git a/common/src/main/java/org/apache/celeborn/common/network/protocol/PushDataHandShake.java b/common/src/main/java/org/apache/celeborn/common/network/protocol/PushDataHandShake.java
index 163fcaeb9..dc8c04816 100644
--- a/common/src/main/java/org/apache/celeborn/common/network/protocol/PushDataHandShake.java
+++ b/common/src/main/java/org/apache/celeborn/common/network/protocol/PushDataHandShake.java
@@ -20,6 +20,7 @@ package org.apache.celeborn.common.network.protocol;
import com.google.common.base.Objects;
import io.netty.buffer.ByteBuf;
+@Deprecated
public final class PushDataHandShake extends RequestMessage {
// 0 for primary, 1 for replica, see PartitionLocation.Mode
public final byte mode;
diff --git a/common/src/main/java/org/apache/celeborn/common/network/protocol/RegionFinish.java b/common/src/main/java/org/apache/celeborn/common/network/protocol/RegionFinish.java
index 62fe18e53..c7f804d84 100644
--- a/common/src/main/java/org/apache/celeborn/common/network/protocol/RegionFinish.java
+++ b/common/src/main/java/org/apache/celeborn/common/network/protocol/RegionFinish.java
@@ -20,6 +20,7 @@ package org.apache.celeborn.common.network.protocol;
import com.google.common.base.Objects;
import io.netty.buffer.ByteBuf;
+@Deprecated
public final class RegionFinish extends RequestMessage {
// 0 for primary, 1 for replica, see PartitionLocation.Mode
diff --git a/common/src/main/java/org/apache/celeborn/common/network/protocol/RegionStart.java b/common/src/main/java/org/apache/celeborn/common/network/protocol/RegionStart.java
index 322029d28..4081c5cf4 100644
--- a/common/src/main/java/org/apache/celeborn/common/network/protocol/RegionStart.java
+++ b/common/src/main/java/org/apache/celeborn/common/network/protocol/RegionStart.java
@@ -20,6 +20,7 @@ package org.apache.celeborn.common.network.protocol;
import com.google.common.base.Objects;
import io.netty.buffer.ByteBuf;
+@Deprecated
public final class RegionStart extends RequestMessage {
// 0 for primary, 1 for replica, see PartitionLocation.Mode
diff --git a/common/src/main/java/org/apache/celeborn/common/network/protocol/TransportMessage.java b/common/src/main/java/org/apache/celeborn/common/network/protocol/TransportMessage.java
index 769eef0d4..965c534d9 100644
--- a/common/src/main/java/org/apache/celeborn/common/network/protocol/TransportMessage.java
+++ b/common/src/main/java/org/apache/celeborn/common/network/protocol/TransportMessage.java
@@ -20,7 +20,10 @@ package org.apache.celeborn.common.network.protocol;
import static org.apache.celeborn.common.protocol.MessageType.BACKLOG_ANNOUNCEMENT_VALUE;
import static org.apache.celeborn.common.protocol.MessageType.BUFFER_STREAM_END_VALUE;
import static org.apache.celeborn.common.protocol.MessageType.OPEN_STREAM_VALUE;
+import static org.apache.celeborn.common.protocol.MessageType.PUSH_DATA_HAND_SHAKE_VALUE;
import static org.apache.celeborn.common.protocol.MessageType.READ_ADD_CREDIT_VALUE;
+import static org.apache.celeborn.common.protocol.MessageType.REGION_FINISH_VALUE;
+import static org.apache.celeborn.common.protocol.MessageType.REGION_START_VALUE;
import static org.apache.celeborn.common.protocol.MessageType.STREAM_HANDLER_VALUE;
import java.io.Serializable;
@@ -36,7 +39,10 @@ import org.apache.celeborn.common.protocol.MessageType;
import org.apache.celeborn.common.protocol.PbBacklogAnnouncement;
import org.apache.celeborn.common.protocol.PbBufferStreamEnd;
import org.apache.celeborn.common.protocol.PbOpenStream;
+import org.apache.celeborn.common.protocol.PbPushDataHandShake;
import org.apache.celeborn.common.protocol.PbReadAddCredit;
+import org.apache.celeborn.common.protocol.PbRegionFinish;
+import org.apache.celeborn.common.protocol.PbRegionStart;
import org.apache.celeborn.common.protocol.PbStreamHandler;
public class TransportMessage implements Serializable {
@@ -76,6 +82,12 @@ public class TransportMessage implements Serializable {
return (T) PbBufferStreamEnd.parseFrom(payload);
case READ_ADD_CREDIT_VALUE:
return (T) PbReadAddCredit.parseFrom(payload);
+ case PUSH_DATA_HAND_SHAKE_VALUE:
+ return (T) PbPushDataHandShake.parseFrom(payload);
+ case REGION_START_VALUE:
+ return (T) PbRegionStart.parseFrom(payload);
+ case REGION_FINISH_VALUE:
+ return (T) PbRegionFinish.parseFrom(payload);
default:
logger.error("Unexpected type {}", type);
}
diff --git a/common/src/main/proto/TransportMessages.proto b/common/src/main/proto/TransportMessages.proto
index e31725edc..fa91bd234 100644
--- a/common/src/main/proto/TransportMessages.proto
+++ b/common/src/main/proto/TransportMessages.proto
@@ -79,6 +79,9 @@ enum MessageType {
BACKLOG_ANNOUNCEMENT = 59;
BUFFER_STREAM_END = 60;
READ_ADD_CREDIT = 61;
+ PUSH_DATA_HAND_SHAKE = 56;
+ REGION_START = 57;
+ REGION_FINISH = 58;
}
message PbStorageInfo {
@@ -527,3 +530,29 @@ message PbReadAddCredit {
int64 streamId = 1;
int32 credit = 2;
}
+
+
+message PbPushDataHandShake {
+ PbPartitionLocation.Mode mode = 1;
+ string shuffleKey = 2;
+ string partitionUniqueId = 3;
+ int32 attemptId = 4;
+ int32 numPartitions = 5;
+ int32 bufferSize = 6;
+}
+
+message PbRegionStart {
+ PbPartitionLocation.Mode mode = 1;
+ string shuffleKey = 2;
+ string partitionUniqueId = 3;
+ int32 attemptId = 4;
+ int32 currentRegionIndex = 5;
+ bool isBroadcast = 6;
+}
+
+message PbRegionFinish {
+ PbPartitionLocation.Mode mode = 1;
+ string shuffleKey = 2;
+ string partitionUniqueId = 3;
+ int32 attemptId = 4;
+}
diff --git a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/PushDataHandler.scala b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/PushDataHandler.scala
index 68facb601..0a201439a 100644
--- a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/PushDataHandler.scala
+++ b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/PushDataHandler.scala
@@ -22,6 +22,7 @@ import java.util.concurrent.{ConcurrentHashMap, ThreadPoolExecutor}
import java.util.concurrent.atomic.{AtomicBoolean, AtomicIntegerArray}
import com.google.common.base.Throwables
+import com.google.protobuf.GeneratedMessageV3
import io.netty.buffer.ByteBuf
import org.apache.celeborn.common.exception.{AlreadyClosedException, CelebornIOException}
@@ -30,10 +31,11 @@ import org.apache.celeborn.common.meta.{DiskStatus, WorkerInfo, WorkerPartitionL
import org.apache.celeborn.common.metrics.source.Source
import org.apache.celeborn.common.network.buffer.{NettyManagedBuffer, NioManagedBuffer}
import org.apache.celeborn.common.network.client.{RpcResponseCallback, TransportClient, TransportClientFactory}
-import org.apache.celeborn.common.network.protocol.{Message, PushData, PushDataHandShake, PushMergedData, RegionFinish, RegionStart, RequestMessage, RpcFailure, RpcRequest, RpcResponse}
+import org.apache.celeborn.common.network.protocol.{Message, PushData, PushDataHandShake, PushMergedData, RegionFinish, RegionStart, RequestMessage, RpcFailure, RpcRequest, RpcResponse, TransportMessage}
import org.apache.celeborn.common.network.protocol.Message.Type
import org.apache.celeborn.common.network.server.BaseMessageHandler
-import org.apache.celeborn.common.protocol.{PartitionLocation, PartitionSplitMode, PartitionType}
+import org.apache.celeborn.common.protocol.{PartitionLocation, PartitionSplitMode, PartitionType, PbPushDataHandShake, PbRegionFinish, PbRegionStart}
+import org.apache.celeborn.common.protocol.PbPartitionLocation.Mode
import org.apache.celeborn.common.protocol.message.StatusCode
import org.apache.celeborn.common.unsafe.Platform
import org.apache.celeborn.common.util.Utils
@@ -798,10 +800,8 @@ class PushDataHandler extends BaseMessageHandler with Logging {
pushData.`type`(),
shuffleKey,
pushData.partitionUniqueId,
- null,
location,
- callback,
- wrappedCallback)) return
+ callback)) return
val fileWriter =
getFileWriterAndCheck(pushData.`type`(), location, isPrimary, callback) match {
@@ -849,45 +849,120 @@ class PushDataHandler extends BaseMessageHandler with Logging {
}
private def handleRpcRequest(client: TransportClient, rpcRequest: RpcRequest): Unit = {
- val msg = Message.decode(rpcRequest.body().nioByteBuffer())
val requestId = rpcRequest.requestId
- val (mode, shuffleKey, partitionUniqueId, checkSplit) = msg match {
- case p: PushDataHandShake => (p.mode, p.shuffleKey, p.partitionUniqueId, true)
- case rs: RegionStart => (rs.mode, rs.shuffleKey, rs.partitionUniqueId, true)
- case rf: RegionFinish => (rf.mode, rf.shuffleKey, rf.partitionUniqueId, false)
- }
+ val (pbMsg, msg, isLegacy, messageType, mode, shuffleKey, partitionUniqueId, checkSplit) =
+ mapPartitionRpcRequest(rpcRequest)
handleCore(
client,
rpcRequest,
requestId,
() =>
handleMapPartitionRpcRequestCore(
- mode,
+ requestId,
+ pbMsg,
msg,
+ isLegacy,
+ messageType,
+ mode,
shuffleKey,
partitionUniqueId,
- requestId,
checkSplit,
new SimpleRpcResponseCallback(
client,
requestId,
shuffleKey)))
+ }
+ private def mapPartitionRpcRequest(rpcRequest: RpcRequest)
+ : Tuple8[GeneratedMessageV3, Message, Boolean, Type, Mode, String, String, Boolean] = {
+ try {
+ val msg = TransportMessage.fromByteBuffer(
+ rpcRequest.body().nioByteBuffer()).getParsedPayload.asInstanceOf[GeneratedMessageV3]
+ msg match {
+ case p: PbPushDataHandShake =>
+ (
+ msg,
+ null,
+ false,
+ Type.PUSH_DATA_HAND_SHAKE,
+ p.getMode,
+ p.getShuffleKey,
+ p.getPartitionUniqueId,
+ true)
+ case rs: PbRegionStart =>
+ (
+ msg,
+ null,
+ false,
+ Type.REGION_START,
+ rs.getMode,
+ rs.getShuffleKey,
+ rs.getPartitionUniqueId,
+ true)
+ case rf: PbRegionFinish =>
+ (
+ msg,
+ null,
+ false,
+ Type.REGION_FINISH,
+ rf.getMode,
+ rf.getShuffleKey,
+ rf.getPartitionUniqueId,
+ false)
+ }
+ } catch {
+ case _: Exception =>
+ val msg = Message.decode(rpcRequest.body().nioByteBuffer())
+ msg match {
+ case p: PushDataHandShake =>
+ (
+ null,
+ msg,
+ true,
+ Type.PUSH_DATA_HAND_SHAKE,
+ Mode.forNumber(p.mode),
+ p.shuffleKey,
+ p.partitionUniqueId,
+ true)
+ case rs: RegionStart =>
+ (
+ null,
+ msg,
+ true,
+ Type.REGION_START,
+ Mode.forNumber(rs.mode),
+ rs.shuffleKey,
+ rs.partitionUniqueId,
+ true)
+ case rf: RegionFinish =>
+ (
+ null,
+ msg,
+ true,
+ Type.REGION_FINISH,
+ Mode.forNumber(rf.mode),
+ rf.shuffleKey,
+ rf.partitionUniqueId,
+ false)
+ }
+ }
}
private def handleMapPartitionRpcRequestCore(
- mode: Byte,
- message: Message,
+ requestId: Long,
+ pbMsg: GeneratedMessageV3,
+ msg: Message,
+ isLegacy: Boolean,
+ messageType: Message.Type,
+ mode: Mode,
shuffleKey: String,
partitionUniqueId: String,
- requestId: Long,
checkSplit: Boolean,
callback: RpcResponseCallback): Unit = {
- val isPrimary = PartitionLocation.getMode(mode) == PartitionLocation.Mode.PRIMARY
- val messageType = message.`type`()
log.debug(
s"requestId:$requestId, pushdata rpc:$messageType, mode:$mode, shuffleKey:$shuffleKey, " +
s"partitionUniqueId:$partitionUniqueId")
+ val isPrimary = mode == Mode.Primary
val (workerSourcePrimary, workerSourceReplica) =
messageType match {
case Type.PUSH_DATA_HAND_SHAKE =>
@@ -924,10 +999,8 @@ class PushDataHandler extends BaseMessageHandler with Logging {
messageType,
shuffleKey,
partitionUniqueId,
- null,
location,
- callback,
- wrappedCallback)) return
+ callback)) return
val fileWriter =
getFileWriterAndCheck(messageType, location, isPrimary, callback) match {
@@ -957,13 +1030,31 @@ class PushDataHandler extends BaseMessageHandler with Logging {
try {
messageType match {
case Type.PUSH_DATA_HAND_SHAKE =>
+ val (numPartitions, bufferSize) =
+ if (isLegacy)
+ (
+ msg.asInstanceOf[PushDataHandShake].numPartitions,
+ msg.asInstanceOf[PushDataHandShake].bufferSize)
+ else
+ (
+ pbMsg.asInstanceOf[PbPushDataHandShake].getNumPartitions,
+ pbMsg.asInstanceOf[PbPushDataHandShake].getBufferSize)
fileWriter.asInstanceOf[MapPartitionFileWriter].pushDataHandShake(
- message.asInstanceOf[PushDataHandShake].numPartitions,
- message.asInstanceOf[PushDataHandShake].bufferSize)
+ numPartitions,
+ bufferSize)
case Type.REGION_START =>
+ val (currentRegionIndex, isBroadcast) =
+ if (isLegacy)
+ (
+ msg.asInstanceOf[RegionStart].currentRegionIndex,
+ msg.asInstanceOf[RegionStart].isBroadcast)
+ else
+ (
+ pbMsg.asInstanceOf[PbRegionStart].getCurrentRegionIndex,
+ Boolean.box(pbMsg.asInstanceOf[PbRegionStart].getIsBroadcast))
fileWriter.asInstanceOf[MapPartitionFileWriter].regionStart(
- message.asInstanceOf[RegionStart].currentRegionIndex,
- message.asInstanceOf[RegionStart].isBroadcast)
+ currentRegionIndex,
+ isBroadcast)
case Type.REGION_FINISH =>
fileWriter.asInstanceOf[MapPartitionFileWriter].regionFinish()
case _ => throw new IllegalArgumentException(s"Not support $messageType yet")
@@ -1039,10 +1130,8 @@ class PushDataHandler extends BaseMessageHandler with Logging {
messageType: Message.Type,
shuffleKey: String,
partitionUniqueId: String,
- body: ByteBuf,
location: PartitionLocation,
- callback: RpcResponseCallback,
- wrappedCallback: RpcResponseCallback): Boolean = {
+ callback: RpcResponseCallback): Boolean = {
if (location == null) {
val msg =
s"Partition location wasn't found for task(shuffle $shuffleKey, uniqueId $partitionUniqueId)."
[incubator-celeborn] 06/07: [CELEBORN-1013] Shutdown master if initialized failed
Posted by ch...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
chengpan pushed a commit to branch branch-0.3
in repository https://gitbox.apache.org/repos/asf/incubator-celeborn.git
commit fd5ecc1e31474adef9d4e18a01bf42d1f1899536
Author: sychen <sy...@ctrip.com>
AuthorDate: Thu Sep 28 19:02:59 2023 +0800
[CELEBORN-1013] Shutdown master if initialized failed
### What changes were proposed in this pull request?
```java
23/09/28 14:48:12,512 ERROR [main] Master: Initialize master failed.
java.net.BindException: Address already in use
at sun.nio.ch.Net.bind0(Native Method)
at sun.nio.ch.Net.bind(Net.java:461)
at sun.nio.ch.Net.bind(Net.java:453)
at sun.nio.ch.ServerSocketChannelImpl.bind(ServerSocketChannelImpl.java:222)
at io.netty.channel.socket.nio.NioServerSocketChannel.doBind(NioServerSocketChannel.java:141)
at io.netty.channel.AbstractChannel$AbstractUnsafe.bind(AbstractChannel.java:562)
at io.netty.channel.DefaultChannelPipeline$HeadContext.bind(DefaultChannelPipeline.java:1334)
```
### Why are the changes needed?
For example, bind's http service port(`celeborn.metrics.master.prometheus.port`) port is occupied and master startup fails, but because the thread started by Raft is not a daemon, the master process still exists.
https://github.com/apache/ratis/blob/d461a01a53e7e130f0ec4143e75b316012137b62/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerImpl.java#L283-L290
### Does this PR introduce _any_ user-facing change?
### How was this patch tested?
Closes #1945 from cxzl25/CELEBORN-1013.
Authored-by: sychen <sy...@ctrip.com>
Signed-off-by: Cheng Pan <ch...@apache.org>
(cherry picked from commit 16bf2aeeaa1767db5f6b676818ce4c9062ed2608)
Signed-off-by: Cheng Pan <ch...@apache.org>
---
.../org/apache/celeborn/service/deploy/master/Master.scala | 10 ++++++++--
1 file changed, 8 insertions(+), 2 deletions(-)
diff --git a/master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala b/master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala
index 4344461c5..737a8a9b5 100644
--- a/master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala
+++ b/master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala
@@ -973,7 +973,13 @@ private[deploy] object Master extends Logging {
def main(args: Array[String]): Unit = {
val conf = new CelebornConf()
val masterArgs = new MasterArguments(args, conf)
- val master = new Master(conf, masterArgs)
- master.initialize()
+ try {
+ val master = new Master(conf, masterArgs)
+ master.initialize()
+ } catch {
+ case e: Throwable =>
+ logError("Initialize master failed.", e)
+ System.exit(-1)
+ }
}
}
[incubator-celeborn] 07/07: [CELEBORN-1014] Output log with bound address and port
Posted by ch...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
chengpan pushed a commit to branch branch-0.3
in repository https://gitbox.apache.org/repos/asf/incubator-celeborn.git
commit f4ffcbeeee816a695850089a11fcaffd75efe095
Author: sychen <sy...@ctrip.com>
AuthorDate: Thu Sep 28 19:07:12 2023 +0800
[CELEBORN-1014] Output log with bound address and port
### What changes were proposed in this pull request?
### Why are the changes needed?
Make it easy for administrators to find the address of the http service bindings.
### Does this PR introduce _any_ user-facing change?
### How was this patch tested?
```
23/09/28 17:10:50,465 INFO [main] HttpServer: master: HttpServer started on port 9983.
```
PR
```
23/09/28 17:28:29,797 INFO [main] HttpServer: master: HttpServer started on clb-3 with port 9983.
```
Closes #1947 from cxzl25/CELEBORN-1014.
Lead-authored-by: sychen <sy...@ctrip.com>
Co-authored-by: Cheng Pan <pa...@gmail.com>
Signed-off-by: Cheng Pan <ch...@apache.org>
(cherry picked from commit 7e944c1a50400931f67f4b42e44239591fe64700)
Signed-off-by: Cheng Pan <ch...@apache.org>
---
.../org/apache/celeborn/common/network/server/TransportServer.java | 2 +-
.../scala/org/apache/celeborn/server/common/http/HttpServer.scala | 5 +++--
2 files changed, 4 insertions(+), 3 deletions(-)
diff --git a/common/src/main/java/org/apache/celeborn/common/network/server/TransportServer.java b/common/src/main/java/org/apache/celeborn/common/network/server/TransportServer.java
index d4214ddb7..73bb38bac 100644
--- a/common/src/main/java/org/apache/celeborn/common/network/server/TransportServer.java
+++ b/common/src/main/java/org/apache/celeborn/common/network/server/TransportServer.java
@@ -116,7 +116,7 @@ public class TransportServer implements Closeable {
channelFuture.syncUninterruptibly();
port = ((InetSocketAddress) channelFuture.channel().localAddress()).getPort();
- logger.debug("Shuffle server started on port: {}", port);
+ logger.debug("Shuffle server started on {}:{}", address.getHostString(), port);
}
protected void initializeChannel(ServerBootstrap bootstrap) {
diff --git a/service/src/main/scala/org/apache/celeborn/server/common/http/HttpServer.scala b/service/src/main/scala/org/apache/celeborn/server/common/http/HttpServer.scala
index f84151333..a45ff109a 100644
--- a/service/src/main/scala/org/apache/celeborn/server/common/http/HttpServer.scala
+++ b/service/src/main/scala/org/apache/celeborn/server/common/http/HttpServer.scala
@@ -50,9 +50,10 @@ class HttpServer(
.channel(classOf[NioServerSocketChannel])
.childHandler(channelInitializer)
- bindFuture = bootstrap.bind(new InetSocketAddress(host, port)).sync
+ val address = new InetSocketAddress(host, port)
+ bindFuture = bootstrap.bind(address).sync
bindFuture.syncUninterruptibly()
- logInfo(s"$role: HttpServer started on port $port.")
+ logInfo(s"$role: HttpServer started on ${address.getHostString}:$port.")
isStarted = true
}