Copilot commented on code in PR #3619:
URL: https://github.com/apache/celeborn/pull/3619#discussion_r3232436662
##########
worker/src/test/scala/org/apache/celeborn/service/deploy/cluster/JavaCppHybridReadWriteTestBase.scala:
##########
@@ -213,31 +218,140 @@ trait JavaCppHybridReadWriteTestBase extends AnyFunSuite
0,
Integer.MAX_VALUE,
metricsCallback)
- var c = inputStream.read()
- var data: Long = 0
- var dataCnt = 0
- while (c != -1) {
- if (c == '-') {
- sums.set(partitionId, sums.get(partitionId) + data)
- data = 0
- dataCnt += 1
- } else {
- assert(c >= '0' && c <= '9')
- data *= 10
- data += c - '0'
+ try {
+ var c = inputStream.read()
+ var data: Long = 0
+ var dataCnt = 0
+ while (c != -1) {
+ if (c == '-') {
+ sums.set(partitionId, sums.get(partitionId) + data)
+ data = 0
+ dataCnt += 1
+ } else {
+ assert(c >= '0' && c <= '9')
+ data *= 10
+ data += c - '0'
+ }
+ c = inputStream.read()
}
- c = inputStream.read()
+ sums.set(partitionId, sums.get(partitionId) + data)
+ println(s"partition $partitionId sum result =
${sums.get(partitionId)}, dataCnt = $dataCnt")
+ } finally {
+ inputStream.close()
}
- sums.set(partitionId, sums.get(partitionId) + data)
- println(s"partition $partitionId sum result = ${sums.get(partitionId)},
dataCnt = $dataCnt")
}
// Verify the sum result.
var lineCount = 0
- for (line <- Source.fromFile(cppResultFile, "utf-8").getLines.toList) {
- val data = line.toLong
- Assert.assertEquals(data, sums.get(lineCount))
- lineCount += 1
+ val source = Source.fromFile(cppResultFile, "utf-8")
+ try {
+ for (line <- source.getLines.toList) {
+ val data = line.toLong
+ Assert.assertEquals(data, sums.get(lineCount))
+ lineCount += 1
+ }
+ } finally {
+ source.close()
+ }
+ Assert.assertEquals(lineCount, numPartitions)
+ lifecycleManager.stop()
+ shuffleClient.shutdown()
+ }
+
+ def testCppMergeWriteJavaRead(codec: CompressionCodec): Unit = {
+ beforeAll()
+ try {
+ runCppMergeWriteJavaRead(codec)
+ } finally {
+ afterAll()
+ }
+ }
+
+ def runCppMergeWriteJavaRead(codec: CompressionCodec): Unit = {
+ val appUniqueId = "test-app"
+ val shuffleId = 0
+ val attemptId = 0
+
+ val clientConf = new CelebornConf()
+ .set(CelebornConf.MASTER_ENDPOINTS.key, s"localhost:$masterPort")
+ .set(CelebornConf.SHUFFLE_COMPRESSION_CODEC.key, codec.name)
+ .set(CelebornConf.CLIENT_PUSH_REPLICATE_ENABLED.key, "true")
+ .set(CelebornConf.CLIENT_PUSH_BUFFER_MAX_SIZE.key, "256K")
+ .set(CelebornConf.READ_LOCAL_SHUFFLE_FILE, false)
+ .set("celeborn.data.io.numConnectionsPerPeer", "1")
+ val lifecycleManager = new LifecycleManager(appUniqueId, clientConf)
+
+ val shuffleClient =
+ new ShuffleClientImpl(appUniqueId, clientConf, UserIdentifier("mock",
"mock"))
+ shuffleClient.setupLifecycleManagerRef(lifecycleManager.self)
+
+ val numMappers = 2
+ val numPartitions = 2
+
+ val cppResultFile = "/tmp/celeborn-cpp-merge-writer-result.txt"
+ val lifecycleManagerHost = lifecycleManager.getHost
+ val lifecycleManagerPort = lifecycleManager.getPort
+ val projectDirectory = new File(new File(".").getAbsolutePath)
+ val cppBinRelativeDirectory = "cpp/build/celeborn/tests/"
+ val cppBinFileName = "cppDataSumWithMergeWriterClient"
+ val cppBinFilePath =
s"$projectDirectory/$cppBinRelativeDirectory/$cppBinFileName"
+ val cppCodec = codec.name()
+ val command = {
+ s"$cppBinFilePath $lifecycleManagerHost $lifecycleManagerPort
$appUniqueId $shuffleId $attemptId $numMappers $numPartitions $cppResultFile
$cppCodec"
+ }
+ println(s"run command: $command")
+ val commandOutput = runCommand(command)
+ println(s"command output: $commandOutput")
+
+ val metricsCallback = new MetricsCallback {
+ override def incBytesRead(bytesWritten: Long): Unit = {}
+ override def incReadTime(time: Long): Unit = {}
+ }
+
+ var sums = new util.ArrayList[Long](numPartitions)
+ for (partitionId <- 0 until numPartitions) {
+ sums.add(0)
+ val inputStream = shuffleClient.readPartition(
+ shuffleId,
+ partitionId,
+ attemptId,
+ 0,
+ 0,
+ Integer.MAX_VALUE,
+ metricsCallback)
+ try {
+ var c = inputStream.read()
+ var data: Long = 0
+ var dataCnt = 0
+ while (c != -1) {
+ if (c == '-') {
+ sums.set(partitionId, sums.get(partitionId) + data)
+ data = 0
+ dataCnt += 1
+ } else {
+ assert(c >= '0' && c <= '9')
+ data *= 10
+ data += c - '0'
+ }
+ c = inputStream.read()
+ }
+ sums.set(partitionId, sums.get(partitionId) + data)
+ println(s"partition $partitionId sum result =
${sums.get(partitionId)}, dataCnt = $dataCnt")
+ } finally {
+ inputStream.close()
+ }
+ }
+
+ var lineCount = 0
+ val source = Source.fromFile(cppResultFile, "utf-8")
+ try {
+ for (line <- source.getLines.toList) {
+ val data = line.toLong
+ Assert.assertEquals(data, sums.get(lineCount))
+ lineCount += 1
+ }
+ } finally {
+ source.close()
}
Assert.assertEquals(lineCount, numPartitions)
lifecycleManager.stop()
Review Comment:
`runCppMergeWriteJavaRead` stops the `lifecycleManager` but does not call
`shuffleClient.shutdown()`. This can leak threads/connections in the test JVM
and can cause the Maven exec run to hang or interfere with subsequent tests.
Add `shuffleClient.shutdown()` as part of the method teardown (ideally in a
`finally` near where `lifecycleManager.stop()` is called, mirroring the
existing non-merge path).
##########
worker/src/test/scala/org/apache/celeborn/service/deploy/cluster/JavaCppHybridReadWriteTestBase.scala:
##########
@@ -213,31 +218,140 @@ trait JavaCppHybridReadWriteTestBase extends AnyFunSuite
0,
Integer.MAX_VALUE,
metricsCallback)
- var c = inputStream.read()
- var data: Long = 0
- var dataCnt = 0
- while (c != -1) {
- if (c == '-') {
- sums.set(partitionId, sums.get(partitionId) + data)
- data = 0
- dataCnt += 1
- } else {
- assert(c >= '0' && c <= '9')
- data *= 10
- data += c - '0'
+ try {
+ var c = inputStream.read()
+ var data: Long = 0
+ var dataCnt = 0
+ while (c != -1) {
+ if (c == '-') {
+ sums.set(partitionId, sums.get(partitionId) + data)
+ data = 0
+ dataCnt += 1
+ } else {
+ assert(c >= '0' && c <= '9')
+ data *= 10
+ data += c - '0'
+ }
+ c = inputStream.read()
}
- c = inputStream.read()
+ sums.set(partitionId, sums.get(partitionId) + data)
+ println(s"partition $partitionId sum result =
${sums.get(partitionId)}, dataCnt = $dataCnt")
+ } finally {
+ inputStream.close()
}
- sums.set(partitionId, sums.get(partitionId) + data)
- println(s"partition $partitionId sum result = ${sums.get(partitionId)},
dataCnt = $dataCnt")
}
// Verify the sum result.
var lineCount = 0
- for (line <- Source.fromFile(cppResultFile, "utf-8").getLines.toList) {
- val data = line.toLong
- Assert.assertEquals(data, sums.get(lineCount))
- lineCount += 1
+ val source = Source.fromFile(cppResultFile, "utf-8")
+ try {
+ for (line <- source.getLines.toList) {
+ val data = line.toLong
+ Assert.assertEquals(data, sums.get(lineCount))
+ lineCount += 1
+ }
+ } finally {
+ source.close()
+ }
+ Assert.assertEquals(lineCount, numPartitions)
+ lifecycleManager.stop()
+ shuffleClient.shutdown()
+ }
+
+ def testCppMergeWriteJavaRead(codec: CompressionCodec): Unit = {
+ beforeAll()
+ try {
+ runCppMergeWriteJavaRead(codec)
+ } finally {
+ afterAll()
+ }
+ }
+
+ def runCppMergeWriteJavaRead(codec: CompressionCodec): Unit = {
+ val appUniqueId = "test-app"
Review Comment:
The test uses a fixed `appUniqueId` and a fixed result file path under
`/tmp`. This can lead to flaky behavior when tests run concurrently (or are
re-run without cleanup), and `/tmp` is not guaranteed to be writable/consistent
across all environments. Prefer generating a unique `appUniqueId` per run
(e.g., include a UUID) and using a per-test temp file (e.g.,
`java.nio.file.Files.createTempFile`) that is cleaned up afterward.
##########
worker/src/test/scala/org/apache/celeborn/service/deploy/cluster/JavaCppHybridReadWriteTestBase.scala:
##########
@@ -213,31 +218,140 @@ trait JavaCppHybridReadWriteTestBase extends AnyFunSuite
0,
Integer.MAX_VALUE,
metricsCallback)
- var c = inputStream.read()
- var data: Long = 0
- var dataCnt = 0
- while (c != -1) {
- if (c == '-') {
- sums.set(partitionId, sums.get(partitionId) + data)
- data = 0
- dataCnt += 1
- } else {
- assert(c >= '0' && c <= '9')
- data *= 10
- data += c - '0'
+ try {
+ var c = inputStream.read()
+ var data: Long = 0
+ var dataCnt = 0
+ while (c != -1) {
+ if (c == '-') {
+ sums.set(partitionId, sums.get(partitionId) + data)
+ data = 0
+ dataCnt += 1
+ } else {
+ assert(c >= '0' && c <= '9')
+ data *= 10
+ data += c - '0'
+ }
+ c = inputStream.read()
}
- c = inputStream.read()
+ sums.set(partitionId, sums.get(partitionId) + data)
+ println(s"partition $partitionId sum result =
${sums.get(partitionId)}, dataCnt = $dataCnt")
+ } finally {
+ inputStream.close()
}
- sums.set(partitionId, sums.get(partitionId) + data)
- println(s"partition $partitionId sum result = ${sums.get(partitionId)},
dataCnt = $dataCnt")
}
// Verify the sum result.
var lineCount = 0
- for (line <- Source.fromFile(cppResultFile, "utf-8").getLines.toList) {
- val data = line.toLong
- Assert.assertEquals(data, sums.get(lineCount))
- lineCount += 1
+ val source = Source.fromFile(cppResultFile, "utf-8")
+ try {
+ for (line <- source.getLines.toList) {
+ val data = line.toLong
+ Assert.assertEquals(data, sums.get(lineCount))
+ lineCount += 1
+ }
+ } finally {
+ source.close()
+ }
+ Assert.assertEquals(lineCount, numPartitions)
+ lifecycleManager.stop()
+ shuffleClient.shutdown()
+ }
+
+ def testCppMergeWriteJavaRead(codec: CompressionCodec): Unit = {
+ beforeAll()
+ try {
+ runCppMergeWriteJavaRead(codec)
+ } finally {
+ afterAll()
+ }
+ }
+
+ def runCppMergeWriteJavaRead(codec: CompressionCodec): Unit = {
+ val appUniqueId = "test-app"
+ val shuffleId = 0
+ val attemptId = 0
+
+ val clientConf = new CelebornConf()
+ .set(CelebornConf.MASTER_ENDPOINTS.key, s"localhost:$masterPort")
+ .set(CelebornConf.SHUFFLE_COMPRESSION_CODEC.key, codec.name)
+ .set(CelebornConf.CLIENT_PUSH_REPLICATE_ENABLED.key, "true")
+ .set(CelebornConf.CLIENT_PUSH_BUFFER_MAX_SIZE.key, "256K")
+ .set(CelebornConf.READ_LOCAL_SHUFFLE_FILE, false)
+ .set("celeborn.data.io.numConnectionsPerPeer", "1")
+ val lifecycleManager = new LifecycleManager(appUniqueId, clientConf)
+
+ val shuffleClient =
+ new ShuffleClientImpl(appUniqueId, clientConf, UserIdentifier("mock",
"mock"))
+ shuffleClient.setupLifecycleManagerRef(lifecycleManager.self)
+
+ val numMappers = 2
+ val numPartitions = 2
+
+ val cppResultFile = "/tmp/celeborn-cpp-merge-writer-result.txt"
Review Comment:
The test uses a fixed `appUniqueId` and a fixed result file path under
`/tmp`. This can lead to flaky behavior when tests run concurrently (or are
re-run without cleanup), and `/tmp` is not guaranteed to be writable/consistent
across all environments. Prefer generating a unique `appUniqueId` per run
(e.g., include a UUID) and using a per-test temp file (e.g.,
`java.nio.file.Files.createTempFile`) that is cleaned up afterward.
##########
cpp/celeborn/tests/DataSumWithMergeWriterClient.cpp:
##########
@@ -0,0 +1,96 @@
+/*
+ * 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.
+ */
+
+#include <folly/init/Init.h>
+#include <cstdio>
Review Comment:
This file uses `assert`, `std::atoi`, and `std::rand` but does not include
the standard headers that define them (`<cassert>` and `<cstdlib>`). Depending
on transitive includes, this can fail to compile on stricter toolchains. Add
the missing includes explicitly.
##########
cpp/celeborn/tests/DataSumWithMergeWriterClient.cpp:
##########
@@ -0,0 +1,96 @@
+/*
+ * 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.
+ */
+
+#include <folly/init/Init.h>
+#include <cstdio>
+#include <fstream>
+#include <iostream>
+
+#include <celeborn/client/ShuffleClient.h>
+
+int main(int argc, char** argv) {
+ folly::init(&argc, &argv, false);
+ assert(argc == 10);
Review Comment:
Using `assert(argc == 10)` for argument validation is fragile because
`assert` is compiled out in release builds, and the program would then read
invalid `argv` entries leading to undefined behavior. Prefer explicit argument
checking that prints a short usage message and returns a non-zero exit code
when `argc` is unexpected.
##########
cpp/celeborn/tests/DataSumWithMergeWriterClient.cpp:
##########
@@ -0,0 +1,96 @@
+/*
+ * 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.
+ */
+
+#include <folly/init/Init.h>
+#include <cstdio>
+#include <fstream>
+#include <iostream>
+
+#include <celeborn/client/ShuffleClient.h>
+
+int main(int argc, char** argv) {
+ folly::init(&argc, &argv, false);
+ assert(argc == 10);
+ std::string lifecycleManagerHost = argv[1];
+ int lifecycleManagerPort = std::atoi(argv[2]);
+ std::string appUniqueId = argv[3];
+ int shuffleId = std::atoi(argv[4]);
+ int attemptId = std::atoi(argv[5]);
+ int numMappers = std::atoi(argv[6]);
+ int numPartitions = std::atoi(argv[7]);
+ std::string resultFile = argv[8];
+ std::string compressCodec = argv[9];
+ std::cout << "lifecycleManagerHost = " << lifecycleManagerHost
+ << ", lifecycleManagerPort = " << lifecycleManagerPort
+ << ", appUniqueId = " << appUniqueId
+ << ", shuffleId = " << shuffleId << ", attemptId = " << attemptId
+ << ", numMappers = " << numMappers
+ << ", numPartitions = " << numPartitions
+ << ", resultFile = " << resultFile
+ << ", compressCodec = " << compressCodec << std::endl;
+
+ auto conf = std::make_shared<celeborn::conf::CelebornConf>();
+ conf->registerProperty(
+ celeborn::conf::CelebornConf::kShuffleCompressionCodec, compressCodec);
+ auto clientEndpoint =
+ std::make_shared<celeborn::client::ShuffleClientEndpoint>(conf);
+ auto shuffleClient = celeborn::client::ShuffleClientImpl::create(
+ appUniqueId, conf, *clientEndpoint);
+ shuffleClient->setupLifecycleManagerRef(
+ lifecycleManagerHost, lifecycleManagerPort);
+
+ long maxData = 1000000;
+ size_t numData = 1000;
+ std::vector<long> result(numPartitions, 0);
+ std::vector<size_t> dataCnt(numPartitions, 0);
+ for (int mapId = 0; mapId < numMappers; mapId++) {
+ for (int partitionId = 0; partitionId < numPartitions; partitionId++) {
+ for (size_t i = 0; i < numData; i++) {
+ int data = std::rand() % maxData;
Review Comment:
This file uses `assert`, `std::atoi`, and `std::rand` but does not include
the standard headers that define them (`<cassert>` and `<cstdlib>`). Depending
on transitive includes, this can fail to compile on stricter toolchains. Add
the missing includes explicitly.
##########
worker/src/test/scala/org/apache/celeborn/service/deploy/cluster/JavaCppHybridReadWriteTestBase.scala:
##########
@@ -213,31 +218,140 @@ trait JavaCppHybridReadWriteTestBase extends AnyFunSuite
0,
Integer.MAX_VALUE,
metricsCallback)
- var c = inputStream.read()
- var data: Long = 0
- var dataCnt = 0
- while (c != -1) {
- if (c == '-') {
- sums.set(partitionId, sums.get(partitionId) + data)
- data = 0
- dataCnt += 1
- } else {
- assert(c >= '0' && c <= '9')
- data *= 10
- data += c - '0'
+ try {
+ var c = inputStream.read()
+ var data: Long = 0
+ var dataCnt = 0
+ while (c != -1) {
+ if (c == '-') {
+ sums.set(partitionId, sums.get(partitionId) + data)
+ data = 0
+ dataCnt += 1
+ } else {
+ assert(c >= '0' && c <= '9')
+ data *= 10
+ data += c - '0'
+ }
+ c = inputStream.read()
}
- c = inputStream.read()
+ sums.set(partitionId, sums.get(partitionId) + data)
+ println(s"partition $partitionId sum result =
${sums.get(partitionId)}, dataCnt = $dataCnt")
+ } finally {
+ inputStream.close()
}
- sums.set(partitionId, sums.get(partitionId) + data)
- println(s"partition $partitionId sum result = ${sums.get(partitionId)},
dataCnt = $dataCnt")
}
// Verify the sum result.
var lineCount = 0
- for (line <- Source.fromFile(cppResultFile, "utf-8").getLines.toList) {
- val data = line.toLong
- Assert.assertEquals(data, sums.get(lineCount))
- lineCount += 1
+ val source = Source.fromFile(cppResultFile, "utf-8")
+ try {
+ for (line <- source.getLines.toList) {
+ val data = line.toLong
+ Assert.assertEquals(data, sums.get(lineCount))
+ lineCount += 1
+ }
+ } finally {
+ source.close()
+ }
+ Assert.assertEquals(lineCount, numPartitions)
+ lifecycleManager.stop()
+ shuffleClient.shutdown()
+ }
+
+ def testCppMergeWriteJavaRead(codec: CompressionCodec): Unit = {
+ beforeAll()
+ try {
+ runCppMergeWriteJavaRead(codec)
+ } finally {
+ afterAll()
+ }
+ }
+
+ def runCppMergeWriteJavaRead(codec: CompressionCodec): Unit = {
+ val appUniqueId = "test-app"
+ val shuffleId = 0
+ val attemptId = 0
+
+ val clientConf = new CelebornConf()
+ .set(CelebornConf.MASTER_ENDPOINTS.key, s"localhost:$masterPort")
+ .set(CelebornConf.SHUFFLE_COMPRESSION_CODEC.key, codec.name)
+ .set(CelebornConf.CLIENT_PUSH_REPLICATE_ENABLED.key, "true")
+ .set(CelebornConf.CLIENT_PUSH_BUFFER_MAX_SIZE.key, "256K")
+ .set(CelebornConf.READ_LOCAL_SHUFFLE_FILE, false)
+ .set("celeborn.data.io.numConnectionsPerPeer", "1")
+ val lifecycleManager = new LifecycleManager(appUniqueId, clientConf)
+
+ val shuffleClient =
+ new ShuffleClientImpl(appUniqueId, clientConf, UserIdentifier("mock",
"mock"))
+ shuffleClient.setupLifecycleManagerRef(lifecycleManager.self)
+
+ val numMappers = 2
+ val numPartitions = 2
+
+ val cppResultFile = "/tmp/celeborn-cpp-merge-writer-result.txt"
+ val lifecycleManagerHost = lifecycleManager.getHost
+ val lifecycleManagerPort = lifecycleManager.getPort
+ val projectDirectory = new File(new File(".").getAbsolutePath)
+ val cppBinRelativeDirectory = "cpp/build/celeborn/tests/"
+ val cppBinFileName = "cppDataSumWithMergeWriterClient"
+ val cppBinFilePath =
s"$projectDirectory/$cppBinRelativeDirectory/$cppBinFileName"
+ val cppCodec = codec.name()
+ val command = {
+ s"$cppBinFilePath $lifecycleManagerHost $lifecycleManagerPort
$appUniqueId $shuffleId $attemptId $numMappers $numPartitions $cppResultFile
$cppCodec"
+ }
+ println(s"run command: $command")
Review Comment:
The command is built as a single shell string with unquoted arguments. This
will break if any path contains spaces (e.g., workspace directory) and can also
make debugging harder when arguments need escaping. Prefer invoking the process
with an argument sequence (e.g., `Seq(cppBinFilePath, lifecycleManagerHost,
...)`) so quoting/escaping is handled correctly.
##########
worker/src/test/scala/org/apache/celeborn/service/deploy/cluster/JavaCppHybridReadWriteTestBase.scala:
##########
@@ -213,31 +218,140 @@ trait JavaCppHybridReadWriteTestBase extends AnyFunSuite
0,
Integer.MAX_VALUE,
metricsCallback)
- var c = inputStream.read()
- var data: Long = 0
- var dataCnt = 0
- while (c != -1) {
- if (c == '-') {
- sums.set(partitionId, sums.get(partitionId) + data)
- data = 0
- dataCnt += 1
- } else {
- assert(c >= '0' && c <= '9')
- data *= 10
- data += c - '0'
+ try {
+ var c = inputStream.read()
+ var data: Long = 0
+ var dataCnt = 0
+ while (c != -1) {
+ if (c == '-') {
+ sums.set(partitionId, sums.get(partitionId) + data)
+ data = 0
+ dataCnt += 1
+ } else {
+ assert(c >= '0' && c <= '9')
+ data *= 10
+ data += c - '0'
+ }
+ c = inputStream.read()
}
- c = inputStream.read()
+ sums.set(partitionId, sums.get(partitionId) + data)
+ println(s"partition $partitionId sum result =
${sums.get(partitionId)}, dataCnt = $dataCnt")
+ } finally {
+ inputStream.close()
}
- sums.set(partitionId, sums.get(partitionId) + data)
- println(s"partition $partitionId sum result = ${sums.get(partitionId)},
dataCnt = $dataCnt")
}
// Verify the sum result.
var lineCount = 0
- for (line <- Source.fromFile(cppResultFile, "utf-8").getLines.toList) {
- val data = line.toLong
- Assert.assertEquals(data, sums.get(lineCount))
- lineCount += 1
+ val source = Source.fromFile(cppResultFile, "utf-8")
+ try {
+ for (line <- source.getLines.toList) {
+ val data = line.toLong
+ Assert.assertEquals(data, sums.get(lineCount))
+ lineCount += 1
+ }
+ } finally {
+ source.close()
+ }
+ Assert.assertEquals(lineCount, numPartitions)
+ lifecycleManager.stop()
+ shuffleClient.shutdown()
+ }
+
+ def testCppMergeWriteJavaRead(codec: CompressionCodec): Unit = {
+ beforeAll()
+ try {
+ runCppMergeWriteJavaRead(codec)
+ } finally {
+ afterAll()
+ }
+ }
+
+ def runCppMergeWriteJavaRead(codec: CompressionCodec): Unit = {
+ val appUniqueId = "test-app"
+ val shuffleId = 0
+ val attemptId = 0
+
+ val clientConf = new CelebornConf()
+ .set(CelebornConf.MASTER_ENDPOINTS.key, s"localhost:$masterPort")
+ .set(CelebornConf.SHUFFLE_COMPRESSION_CODEC.key, codec.name)
+ .set(CelebornConf.CLIENT_PUSH_REPLICATE_ENABLED.key, "true")
+ .set(CelebornConf.CLIENT_PUSH_BUFFER_MAX_SIZE.key, "256K")
+ .set(CelebornConf.READ_LOCAL_SHUFFLE_FILE, false)
+ .set("celeborn.data.io.numConnectionsPerPeer", "1")
+ val lifecycleManager = new LifecycleManager(appUniqueId, clientConf)
+
+ val shuffleClient =
+ new ShuffleClientImpl(appUniqueId, clientConf, UserIdentifier("mock",
"mock"))
+ shuffleClient.setupLifecycleManagerRef(lifecycleManager.self)
+
+ val numMappers = 2
+ val numPartitions = 2
+
+ val cppResultFile = "/tmp/celeborn-cpp-merge-writer-result.txt"
+ val lifecycleManagerHost = lifecycleManager.getHost
+ val lifecycleManagerPort = lifecycleManager.getPort
+ val projectDirectory = new File(new File(".").getAbsolutePath)
+ val cppBinRelativeDirectory = "cpp/build/celeborn/tests/"
+ val cppBinFileName = "cppDataSumWithMergeWriterClient"
+ val cppBinFilePath =
s"$projectDirectory/$cppBinRelativeDirectory/$cppBinFileName"
+ val cppCodec = codec.name()
+ val command = {
+ s"$cppBinFilePath $lifecycleManagerHost $lifecycleManagerPort
$appUniqueId $shuffleId $attemptId $numMappers $numPartitions $cppResultFile
$cppCodec"
+ }
+ println(s"run command: $command")
+ val commandOutput = runCommand(command)
+ println(s"command output: $commandOutput")
+
+ val metricsCallback = new MetricsCallback {
+ override def incBytesRead(bytesWritten: Long): Unit = {}
+ override def incReadTime(time: Long): Unit = {}
+ }
+
+ var sums = new util.ArrayList[Long](numPartitions)
+ for (partitionId <- 0 until numPartitions) {
+ sums.add(0)
+ val inputStream = shuffleClient.readPartition(
+ shuffleId,
+ partitionId,
+ attemptId,
+ 0,
+ 0,
+ Integer.MAX_VALUE,
+ metricsCallback)
+ try {
+ var c = inputStream.read()
+ var data: Long = 0
+ var dataCnt = 0
+ while (c != -1) {
+ if (c == '-') {
+ sums.set(partitionId, sums.get(partitionId) + data)
+ data = 0
+ dataCnt += 1
+ } else {
+ assert(c >= '0' && c <= '9')
+ data *= 10
+ data += c - '0'
+ }
+ c = inputStream.read()
+ }
+ sums.set(partitionId, sums.get(partitionId) + data)
+ println(s"partition $partitionId sum result =
${sums.get(partitionId)}, dataCnt = $dataCnt")
+ } finally {
+ inputStream.close()
+ }
+ }
+
+ var lineCount = 0
+ val source = Source.fromFile(cppResultFile, "utf-8")
+ try {
+ for (line <- source.getLines.toList) {
Review Comment:
`getLines.toList` reads the full file into memory before iterating. Since
this is streamed input, iterating directly over `source.getLines()` avoids
unnecessary allocation and is simpler. (This is likely small in tests, so it’s
a minor improvement.)
##########
cpp/celeborn/tests/DataSumWithMergeWriterClient.cpp:
##########
@@ -0,0 +1,96 @@
+/*
+ * 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.
+ */
+
+#include <folly/init/Init.h>
+#include <cstdio>
+#include <fstream>
+#include <iostream>
+
+#include <celeborn/client/ShuffleClient.h>
+
+int main(int argc, char** argv) {
+ folly::init(&argc, &argv, false);
+ assert(argc == 10);
+ std::string lifecycleManagerHost = argv[1];
+ int lifecycleManagerPort = std::atoi(argv[2]);
Review Comment:
This file uses `assert`, `std::atoi`, and `std::rand` but does not include
the standard headers that define them (`<cassert>` and `<cstdlib>`). Depending
on transitive includes, this can fail to compile on stricter toolchains. Add
the missing includes explicitly.
##########
cpp/celeborn/tests/CMakeLists.txt:
##########
@@ -61,3 +61,28 @@ target_link_libraries(
add_executable(cppDataSumWithWriterClient DataSumWithWriterClient.cpp)
target_link_libraries(cppDataSumWithWriterClient dataSumWithWriterClient)
+
+add_library(
+ dataSumWithMergeWriterClient
+ DataSumWithMergeWriterClient.cpp)
Review Comment:
`DataSumWithMergeWriterClient.cpp` is compiled both into a library and again
into the executable. This increases build time and is awkward because the
source contains `main()`, which generally should not live inside a library
target. Prefer either: (1) link the executable directly to the needed dependent
libraries without creating a separate `add_library`, or (2) create an OBJECT
library for shared sources (excluding `main`) and link that object library into
the executable.
##########
cpp/celeborn/tests/CMakeLists.txt:
##########
@@ -61,3 +61,28 @@ target_link_libraries(
add_executable(cppDataSumWithWriterClient DataSumWithWriterClient.cpp)
target_link_libraries(cppDataSumWithWriterClient dataSumWithWriterClient)
+
+add_library(
+ dataSumWithMergeWriterClient
+ DataSumWithMergeWriterClient.cpp)
+
+target_link_libraries(
+ dataSumWithMergeWriterClient
+ memory
+ utils
+ conf
+ proto
+ network
+ protocol
+ client
+ ${WANGLE}
+ ${FIZZ}
+ ${LIBSODIUM_LIBRARY}
+ ${FOLLY_WITH_DEPENDENCIES}
+ ${GLOG}
+ ${GFLAGS_LIBRARIES}
+)
+
+add_executable(cppDataSumWithMergeWriterClient
DataSumWithMergeWriterClient.cpp)
+
+target_link_libraries(cppDataSumWithMergeWriterClient
dataSumWithMergeWriterClient)
Review Comment:
`DataSumWithMergeWriterClient.cpp` is compiled both into a library and again
into the executable. This increases build time and is awkward because the
source contains `main()`, which generally should not live inside a library
target. Prefer either: (1) link the executable directly to the needed dependent
libraries without creating a separate `add_library`, or (2) create an OBJECT
library for shared sources (excluding `main`) and link that object library into
the executable.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
