You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@uniffle.apache.org by "smallzhongfeng (via GitHub)" <gi...@apache.org> on 2023/03/03 09:00:26 UTC
[GitHub] [incubator-uniffle] smallzhongfeng opened a new pull request, #680: [#678] improvement: Write hdfs files asynchronously when `detectStorage`
smallzhongfeng opened a new pull request, #680:
URL: https://github.com/apache/incubator-uniffle/pull/680
<!--
1. Title: [#<issue>] <type>(<scope>): <subject>
Examples:
- "[#123] feat(operator): support xxx"
- "[#233] fix: check null before access result in xxx"
- "[MINOR] refactor: fix typo in variable name"
- "[MINOR] docs: fix typo in README"
- "[#255] test: fix flaky test NameOfTheTest"
Reference: https://www.conventionalcommits.org/en/v1.0.0/
2. Contributor guidelines:
https://github.com/apache/incubator-uniffle/blob/master/CONTRIBUTING.md
3. If the PR is unfinished, please mark this PR as draft.
-->
### What changes were proposed in this pull request?
Write files to hdfs asynchronously and reduce the time of detectStorage
### Why are the changes needed?
Fix: #678
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Origin uts.
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] smallzhongfeng commented on pull request #680: [#678] improvement: Write hdfs files asynchronously when `detectStorage`
Posted by "smallzhongfeng (via GitHub)" <gi...@apache.org>.
smallzhongfeng commented on PR #680:
URL: https://github.com/apache/incubator-uniffle/pull/680#issuecomment-1455918684
Thanks all! @advancedxy @xianjingfeng @jerqi
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] smallzhongfeng closed pull request #680: [#678] improvement: Write hdfs files asynchronously when `detectStorage`
Posted by "smallzhongfeng (via GitHub)" <gi...@apache.org>.
smallzhongfeng closed pull request #680: [#678] improvement: Write hdfs files asynchronously when `detectStorage`
URL: https://github.com/apache/incubator-uniffle/pull/680
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] xianjingfeng commented on a diff in pull request #680: [#678] improvement: Write hdfs files asynchronously when `detectStorage`
Posted by "xianjingfeng (via GitHub)" <gi...@apache.org>.
xianjingfeng commented on code in PR #680:
URL: https://github.com/apache/incubator-uniffle/pull/680#discussion_r1125425259
##########
coordinator/src/main/java/org/apache/uniffle/coordinator/strategy/storage/AbstractSelectStorageStrategy.java:
##########
@@ -76,6 +92,50 @@ public void readAndWriteHdfsStorage(FileSystem fs, Path testPath,
}
}
+ @Override
+ public void detectStorage() {
+ uris = Lists.newCopyOnWriteArrayList(remoteStoragePathRankValue.entrySet());
+ if (remoteStoragePathRankValue.size() > 1) {
+ CountDownLatch countDownLatch = new CountDownLatch(uris.size());
+ uris.parallelStream().forEach(uri -> {
+ if (uri.getKey().startsWith(ApplicationManager.getPathSchema().get(0))) {
+ Path remotePath = new Path(uri.getKey());
+ String rssTest = uri.getKey() + "/rssTest-" + getCoordinatorId()
+ + Thread.currentThread().getName();
+ Path testPath = new Path(rssTest);
+ RankValue rankValue = remoteStoragePathRankValue.get(uri.getKey());
+ rankValue.setHealthy(new AtomicBoolean(true));
+ long startWriteTime = System.currentTimeMillis();
+ try {
+ FileSystem fs = HadoopFilesystemProvider.getFilesystem(remotePath, hdfsConf);
+ for (int j = 0; j < readAndWriteTimes; j++) {
Review Comment:
In the original logic in `AppBalanceSelectStorageStrategy`, readAndWriteTimes should be 1. I am not sure if we should keep the original logic.
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] advancedxy commented on a diff in pull request #680: [#678] improvement: Write hdfs files asynchronously when `detectStorage`
Posted by "advancedxy (via GitHub)" <gi...@apache.org>.
advancedxy commented on code in PR #680:
URL: https://github.com/apache/incubator-uniffle/pull/680#discussion_r1126051054
##########
coordinator/src/test/java/org/apache/uniffle/coordinator/strategy/storage/LowestIOSampleCostSelectStorageStrategyTest.java:
##########
@@ -90,6 +91,7 @@ public void selectStorageTest() throws Exception {
assertEquals(0, applicationManager.getRemoteStoragePathRankValue().get(remoteStorage2).getAppNum().get());
// compare with two remote path
+ final Configuration configuration = new Configuration();
Review Comment:
Seems this line is not needed anymore?
I didn't see the var `configuration` is used in this method anywhere.
##########
coordinator/src/main/java/org/apache/uniffle/coordinator/strategy/storage/AbstractSelectStorageStrategy.java:
##########
@@ -76,6 +97,79 @@ public void readAndWriteHdfsStorage(FileSystem fs, Path testPath,
}
}
+ @Override
+ public void detectStorage() {
+ uris = Lists.newCopyOnWriteArrayList(remoteStoragePathRankValue.entrySet());
+ if (remoteStoragePathRankValue.size() > 1) {
+ CountDownLatch countDownLatch = new CountDownLatch(uris.size());
+ uris.parallelStream().forEach(uri -> {
+ if (uri.getKey().startsWith(ApplicationManager.getPathSchema().get(0))) {
+ Path remotePath = new Path(uri.getKey());
+ String rssTest = uri.getKey() + "/rssTest-" + getCoordinatorId()
+ + Thread.currentThread().getName();
+ Path testPath = new Path(rssTest);
+ RankValue rankValue = remoteStoragePathRankValue.get(uri.getKey());
+ rankValue.setHealthy(new AtomicBoolean(true));
+ long startWriteTime = System.currentTimeMillis();
+ try {
+ FileSystem fs = HadoopFilesystemProvider.getFilesystem(remotePath, hdfsConf);
+ for (int j = 0; j < readAndWriteTimes(conf); j++) {
+ readAndWriteHdfsStorage(fs, testPath, uri.getKey(), rankValue);
+ }
+ } catch (Exception e) {
+ LOG.error("Storage read and write error, we will not use this remote path {}.", uri, e);
+ rankValue.setHealthy(new AtomicBoolean(false));
+ } finally {
+ sortPathByRankValue(uri.getKey(), rssTest, startWriteTime);
+ }
+ countDownLatch.countDown();
+ }
+ });
+ try {
+ countDownLatch.await();
+ } catch (InterruptedException e) {
+ LOG.error("Failed to detectStorage!");
+ }
+ }
+ }
+
+ @VisibleForTesting
+ public void sortPathByRankValue(
+ String path, String testPath, long startWrite) {
+ RankValue rankValue = remoteStoragePathRankValue.get(path);
+ try {
+ FileSystem fs = HadoopFilesystemProvider.getFilesystem(new Path(path), hdfsConf);
+ fs.delete(new Path(testPath), true);
+ if (rankValue.getHealthy().get()) {
+ rankValue.setCostTime(new AtomicLong(System.currentTimeMillis() - startWrite));
+ }
+ } catch (Exception e) {
+ rankValue.setCostTime(new AtomicLong(Long.MAX_VALUE));
+ LOG.error("Failed to sort, we will not use this remote path {}.", path, e);
+ }
+
+ if (this.getComparator() != null) {
+ uris = Lists.newCopyOnWriteArrayList(
+ remoteStoragePathRankValue.entrySet()).stream().filter(
+ Objects::nonNull).sorted(this.getComparator()).collect(Collectors.toList());
+ } else {
+ uris = Lists.newCopyOnWriteArrayList(
+ remoteStoragePathRankValue.entrySet()).stream().filter(
+ Objects::nonNull).collect(Collectors.toList());
+ }
+ LOG.info("The sorted remote path list is: {}", uris);
+ }
+
+ public int readAndWriteTimes(CoordinatorConf conf) {
+ return 1;
+ }
+
+ /**
+ * Different strategies will have different sorting methods of remote paths
+ * @return A comparator is to calculate the RankValue
+ */
+ abstract Comparator<Map.Entry<String, RankValue>> getComparator();
Review Comment:
Let's make this method as
```
protected Comparator<Map.Entry<String, RankValue>> getComparator() {
return null;
}
```
? So `AppBalanceSelectorStorageStrategy` doesn't need to override this.
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] smallzhongfeng commented on a diff in pull request #680: [#678] improvement: Write hdfs files asynchronously when `detectStorage`
Posted by "smallzhongfeng (via GitHub)" <gi...@apache.org>.
smallzhongfeng commented on code in PR #680:
URL: https://github.com/apache/incubator-uniffle/pull/680#discussion_r1125453742
##########
coordinator/src/main/java/org/apache/uniffle/coordinator/strategy/storage/AbstractSelectStorageStrategy.java:
##########
@@ -76,6 +92,50 @@ public void readAndWriteHdfsStorage(FileSystem fs, Path testPath,
}
}
+ @Override
+ public void detectStorage() {
+ uris = Lists.newCopyOnWriteArrayList(remoteStoragePathRankValue.entrySet());
+ if (remoteStoragePathRankValue.size() > 1) {
+ CountDownLatch countDownLatch = new CountDownLatch(uris.size());
+ uris.parallelStream().forEach(uri -> {
+ if (uri.getKey().startsWith(ApplicationManager.getPathSchema().get(0))) {
+ Path remotePath = new Path(uri.getKey());
+ String rssTest = uri.getKey() + "/rssTest-" + getCoordinatorId()
+ + Thread.currentThread().getName();
+ Path testPath = new Path(rssTest);
+ RankValue rankValue = remoteStoragePathRankValue.get(uri.getKey());
+ rankValue.setHealthy(new AtomicBoolean(true));
+ long startWriteTime = System.currentTimeMillis();
+ try {
+ FileSystem fs = HadoopFilesystemProvider.getFilesystem(remotePath, hdfsConf);
+ for (int j = 0; j < readAndWriteTimes; j++) {
+ readAndWriteHdfsStorage(fs, testPath, uri.getKey(), rankValue);
+ }
+ } catch (Exception e) {
+ LOG.error("Storage read and write error, we will not use this remote path {}.", uri, e);
+ rankValue.setHealthy(new AtomicBoolean(false));
+ } finally {
+ if (this instanceof LowestIOSampleCostSelectStorageStrategy) {
+ ((LowestIOSampleCostSelectStorageStrategy) this)
+ .sortPathByRankValue(uri.getKey(), rssTest, startWriteTime, hdfsConf);
+ } else if (this instanceof AppBalanceSelectStorageStrategy) {
+ ((AppBalanceSelectStorageStrategy) this)
+ .sortPathByRankValue(uri.getKey(), rssTest, hdfsConf);
Review Comment:
I don't quite understand the meaning of the second point, why the getComparator of AppBalanceSelectStorageStrategy returns null? @xianjingfeng
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] smallzhongfeng commented on a diff in pull request #680: [#678] improvement: Write hdfs files asynchronously when `detectStorage`
Posted by "smallzhongfeng (via GitHub)" <gi...@apache.org>.
smallzhongfeng commented on code in PR #680:
URL: https://github.com/apache/incubator-uniffle/pull/680#discussion_r1125482860
##########
coordinator/src/main/java/org/apache/uniffle/coordinator/strategy/storage/AbstractSelectStorageStrategy.java:
##########
@@ -76,6 +92,50 @@ public void readAndWriteHdfsStorage(FileSystem fs, Path testPath,
}
}
+ @Override
+ public void detectStorage() {
+ uris = Lists.newCopyOnWriteArrayList(remoteStoragePathRankValue.entrySet());
+ if (remoteStoragePathRankValue.size() > 1) {
+ CountDownLatch countDownLatch = new CountDownLatch(uris.size());
+ uris.parallelStream().forEach(uri -> {
+ if (uri.getKey().startsWith(ApplicationManager.getPathSchema().get(0))) {
+ Path remotePath = new Path(uri.getKey());
+ String rssTest = uri.getKey() + "/rssTest-" + getCoordinatorId()
+ + Thread.currentThread().getName();
+ Path testPath = new Path(rssTest);
+ RankValue rankValue = remoteStoragePathRankValue.get(uri.getKey());
+ rankValue.setHealthy(new AtomicBoolean(true));
+ long startWriteTime = System.currentTimeMillis();
+ try {
+ FileSystem fs = HadoopFilesystemProvider.getFilesystem(remotePath, hdfsConf);
+ for (int j = 0; j < readAndWriteTimes; j++) {
+ readAndWriteHdfsStorage(fs, testPath, uri.getKey(), rankValue);
+ }
+ } catch (Exception e) {
+ LOG.error("Storage read and write error, we will not use this remote path {}.", uri, e);
+ rankValue.setHealthy(new AtomicBoolean(false));
+ } finally {
+ if (this instanceof LowestIOSampleCostSelectStorageStrategy) {
+ ((LowestIOSampleCostSelectStorageStrategy) this)
+ .sortPathByRankValue(uri.getKey(), rssTest, startWriteTime, hdfsConf);
+ } else if (this instanceof AppBalanceSelectStorageStrategy) {
+ ((AppBalanceSelectStorageStrategy) this)
+ .sortPathByRankValue(uri.getKey(), rssTest, hdfsConf);
+ } else {
+ LOG.error("Failed to sort path by detectStorage!");
+ }
+ countDownLatch.countDown();
Review Comment:
Updated.
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on pull request #680: [#678] improvement: Write hdfs files asynchronously when `detectStorage`
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on PR #680:
URL: https://github.com/apache/incubator-uniffle/pull/680#issuecomment-1453223203
We should avoid multple threads operate the same file.
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] smallzhongfeng commented on pull request #680: [#678] improvement: Write hdfs files asynchronously when `detectStorage`
Posted by "smallzhongfeng (via GitHub)" <gi...@apache.org>.
smallzhongfeng commented on PR #680:
URL: https://github.com/apache/incubator-uniffle/pull/680#issuecomment-1453670018
This doesn't seem to be caused by this pr.
```
Errors:
Error: QuorumTest.case2:337->registerShuffleServer:281 » StatusRuntime DEADLINE_EXCEE...
```
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] smallzhongfeng commented on a diff in pull request #680: [#678] improvement: Write hdfs files asynchronously when `detectStorage`
Posted by "smallzhongfeng (via GitHub)" <gi...@apache.org>.
smallzhongfeng commented on code in PR #680:
URL: https://github.com/apache/incubator-uniffle/pull/680#discussion_r1125483941
##########
coordinator/src/main/java/org/apache/uniffle/coordinator/strategy/storage/AbstractSelectStorageStrategy.java:
##########
@@ -76,6 +92,50 @@ public void readAndWriteHdfsStorage(FileSystem fs, Path testPath,
}
}
+ @Override
+ public void detectStorage() {
+ uris = Lists.newCopyOnWriteArrayList(remoteStoragePathRankValue.entrySet());
+ if (remoteStoragePathRankValue.size() > 1) {
+ CountDownLatch countDownLatch = new CountDownLatch(uris.size());
+ uris.parallelStream().forEach(uri -> {
+ if (uri.getKey().startsWith(ApplicationManager.getPathSchema().get(0))) {
+ Path remotePath = new Path(uri.getKey());
+ String rssTest = uri.getKey() + "/rssTest-" + getCoordinatorId()
+ + Thread.currentThread().getName();
+ Path testPath = new Path(rssTest);
+ RankValue rankValue = remoteStoragePathRankValue.get(uri.getKey());
+ rankValue.setHealthy(new AtomicBoolean(true));
+ long startWriteTime = System.currentTimeMillis();
+ try {
+ FileSystem fs = HadoopFilesystemProvider.getFilesystem(remotePath, hdfsConf);
+ for (int j = 0; j < readAndWriteTimes; j++) {
+ readAndWriteHdfsStorage(fs, testPath, uri.getKey(), rankValue);
+ }
+ } catch (Exception e) {
+ LOG.error("Storage read and write error, we will not use this remote path {}.", uri, e);
+ rankValue.setHealthy(new AtomicBoolean(false));
+ } finally {
+ if (this instanceof LowestIOSampleCostSelectStorageStrategy) {
+ ((LowestIOSampleCostSelectStorageStrategy) this)
+ .sortPathByRankValue(uri.getKey(), rssTest, startWriteTime, hdfsConf);
+ } else if (this instanceof AppBalanceSelectStorageStrategy) {
+ ((AppBalanceSelectStorageStrategy) this)
+ .sortPathByRankValue(uri.getKey(), rssTest, hdfsConf);
Review Comment:
Do you mean like this https://github.com/apache/incubator-uniffle/pull/680/commits/fd752501e6165d6f59e9b88c5d52d449a9a154bf
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] smallzhongfeng commented on pull request #680: [#678] improvement: Write hdfs files asynchronously when `detectStorage`
Posted by "smallzhongfeng (via GitHub)" <gi...@apache.org>.
smallzhongfeng commented on PR #680:
URL: https://github.com/apache/incubator-uniffle/pull/680#issuecomment-1453256362
> We should avoid multple threads operate the same file.
It is the same file, but it is written to a different path.
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] smallzhongfeng commented on a diff in pull request #680: [#678] improvement: Write hdfs files asynchronously when `detectStorage`
Posted by "smallzhongfeng (via GitHub)" <gi...@apache.org>.
smallzhongfeng commented on code in PR #680:
URL: https://github.com/apache/incubator-uniffle/pull/680#discussion_r1125482256
##########
coordinator/src/main/java/org/apache/uniffle/coordinator/strategy/storage/AbstractSelectStorageStrategy.java:
##########
@@ -76,6 +92,50 @@ public void readAndWriteHdfsStorage(FileSystem fs, Path testPath,
}
}
+ @Override
+ public void detectStorage() {
+ uris = Lists.newCopyOnWriteArrayList(remoteStoragePathRankValue.entrySet());
+ if (remoteStoragePathRankValue.size() > 1) {
+ CountDownLatch countDownLatch = new CountDownLatch(uris.size());
+ uris.parallelStream().forEach(uri -> {
+ if (uri.getKey().startsWith(ApplicationManager.getPathSchema().get(0))) {
+ Path remotePath = new Path(uri.getKey());
+ String rssTest = uri.getKey() + "/rssTest-" + getCoordinatorId()
+ + Thread.currentThread().getName();
+ Path testPath = new Path(rssTest);
+ RankValue rankValue = remoteStoragePathRankValue.get(uri.getKey());
+ rankValue.setHealthy(new AtomicBoolean(true));
+ long startWriteTime = System.currentTimeMillis();
+ try {
+ FileSystem fs = HadoopFilesystemProvider.getFilesystem(remotePath, hdfsConf);
+ for (int j = 0; j < readAndWriteTimes; j++) {
Review Comment:
It's my negligence.
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #680: [#678] improvement: Write hdfs files asynchronously when `detectStorage`
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #680:
URL: https://github.com/apache/incubator-uniffle/pull/680#discussion_r1125604230
##########
coordinator/src/main/java/org/apache/uniffle/coordinator/strategy/storage/AbstractSelectStorageStrategy.java:
##########
@@ -76,6 +92,50 @@ public void readAndWriteHdfsStorage(FileSystem fs, Path testPath,
}
}
+ @Override
+ public void detectStorage() {
+ uris = Lists.newCopyOnWriteArrayList(remoteStoragePathRankValue.entrySet());
+ if (remoteStoragePathRankValue.size() > 1) {
+ CountDownLatch countDownLatch = new CountDownLatch(uris.size());
+ uris.parallelStream().forEach(uri -> {
+ if (uri.getKey().startsWith(ApplicationManager.getPathSchema().get(0))) {
+ Path remotePath = new Path(uri.getKey());
+ String rssTest = uri.getKey() + "/rssTest-" + getCoordinatorId()
+ + Thread.currentThread().getName();
+ Path testPath = new Path(rssTest);
+ RankValue rankValue = remoteStoragePathRankValue.get(uri.getKey());
+ rankValue.setHealthy(new AtomicBoolean(true));
+ long startWriteTime = System.currentTimeMillis();
+ try {
+ FileSystem fs = HadoopFilesystemProvider.getFilesystem(remotePath, hdfsConf);
+ for (int j = 0; j < readAndWriteTimes; j++) {
Review Comment:
Ok for me.
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] smallzhongfeng commented on a diff in pull request #680: [#678] improvement: Write hdfs files asynchronously when `detectStorage`
Posted by "smallzhongfeng (via GitHub)" <gi...@apache.org>.
smallzhongfeng commented on code in PR #680:
URL: https://github.com/apache/incubator-uniffle/pull/680#discussion_r1125375094
##########
coordinator/src/main/java/org/apache/uniffle/coordinator/strategy/storage/AppBalanceSelectStorageStrategy.java:
##########
@@ -84,23 +85,35 @@ public void sortPathByRankValue(String path, String test) {
public void detectStorage() {
uris = Lists.newCopyOnWriteArrayList(remoteStoragePathRankValue.entrySet());
if (remoteStoragePathRankValue.size() > 1) {
+ CountDownLatch countDownLatch = new CountDownLatch(uris.size());
for (Map.Entry<String, RankValue> uri : uris) {
- if (uri.getKey().startsWith(ApplicationManager.getPathSchema().get(0))) {
- RankValue rankValue = remoteStoragePathRankValue.get(uri.getKey());
- rankValue.setHealthy(new AtomicBoolean(true));
- Path remotePath = new Path(uri.getKey());
- String rssTest = uri.getKey() + "/rssTest-" + getCoordinatorId();
- Path testPath = new Path(rssTest);
- try {
- FileSystem fs = HadoopFilesystemProvider.getFilesystem(remotePath, hdfsConf);
- readAndWriteHdfsStorage(fs, testPath, uri.getKey(), rankValue);
- } catch (Exception e) {
- rankValue.setHealthy(new AtomicBoolean(false));
- LOG.error("Storage read and write error, we will not use this remote path {}.", uri, e);
- } finally {
- sortPathByRankValue(uri.getKey(), rssTest);
+ Thread detectThread = new Thread(() -> {
Review Comment:
I use `parallelStream`, the two should be similar. right?
##########
coordinator/src/main/java/org/apache/uniffle/coordinator/strategy/storage/LowestIOSampleCostSelectStorageStrategy.java:
##########
@@ -99,26 +100,38 @@ public void sortPathByRankValue(
public void detectStorage() {
Review Comment:
Updated
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] xianjingfeng commented on a diff in pull request #680: [#678] improvement: Write hdfs files asynchronously when `detectStorage`
Posted by "xianjingfeng (via GitHub)" <gi...@apache.org>.
xianjingfeng commented on code in PR #680:
URL: https://github.com/apache/incubator-uniffle/pull/680#discussion_r1125423521
##########
coordinator/src/main/java/org/apache/uniffle/coordinator/strategy/storage/AbstractSelectStorageStrategy.java:
##########
@@ -76,6 +92,50 @@ public void readAndWriteHdfsStorage(FileSystem fs, Path testPath,
}
}
+ @Override
+ public void detectStorage() {
+ uris = Lists.newCopyOnWriteArrayList(remoteStoragePathRankValue.entrySet());
+ if (remoteStoragePathRankValue.size() > 1) {
+ CountDownLatch countDownLatch = new CountDownLatch(uris.size());
+ uris.parallelStream().forEach(uri -> {
+ if (uri.getKey().startsWith(ApplicationManager.getPathSchema().get(0))) {
+ Path remotePath = new Path(uri.getKey());
+ String rssTest = uri.getKey() + "/rssTest-" + getCoordinatorId()
+ + Thread.currentThread().getName();
+ Path testPath = new Path(rssTest);
+ RankValue rankValue = remoteStoragePathRankValue.get(uri.getKey());
+ rankValue.setHealthy(new AtomicBoolean(true));
+ long startWriteTime = System.currentTimeMillis();
+ try {
+ FileSystem fs = HadoopFilesystemProvider.getFilesystem(remotePath, hdfsConf);
+ for (int j = 0; j < readAndWriteTimes; j++) {
+ readAndWriteHdfsStorage(fs, testPath, uri.getKey(), rankValue);
+ }
+ } catch (Exception e) {
+ LOG.error("Storage read and write error, we will not use this remote path {}.", uri, e);
+ rankValue.setHealthy(new AtomicBoolean(false));
+ } finally {
+ if (this instanceof LowestIOSampleCostSelectStorageStrategy) {
+ ((LowestIOSampleCostSelectStorageStrategy) this)
+ .sortPathByRankValue(uri.getKey(), rssTest, startWriteTime, hdfsConf);
+ } else if (this instanceof AppBalanceSelectStorageStrategy) {
+ ((AppBalanceSelectStorageStrategy) this)
+ .sortPathByRankValue(uri.getKey(), rssTest, hdfsConf);
Review Comment:
I think `sortPathByRankValue` can be merged too.
1. We can add an interface named `getComparator` to SelectStorageStrategy and move `sortPathByRankValue` to `AbstractSelectStorageStrategy`.
2. Return null in the overrided method in `AppBalanceSelectStorageStrategy` and return a comparator in `LowestIOSampleCostSelectStorageStrategy`
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] advancedxy commented on a diff in pull request #680: [#678] improvement: Write hdfs files asynchronously when `detectStorage`
Posted by "advancedxy (via GitHub)" <gi...@apache.org>.
advancedxy commented on code in PR #680:
URL: https://github.com/apache/incubator-uniffle/pull/680#discussion_r1125659829
##########
coordinator/src/main/java/org/apache/uniffle/coordinator/strategy/storage/SelectStorageStrategy.java:
##########
@@ -17,11 +17,15 @@
package org.apache.uniffle.coordinator.strategy.storage;
+import java.util.Comparator;
+
import org.apache.uniffle.common.RemoteStorageInfo;
public interface SelectStorageStrategy {
void detectStorage();
RemoteStorageInfo pickStorage(String appId);
+
+ Comparator getComparator();
Review Comment:
I believe the getComparator is more appropriate to be in the abstract class.
##########
coordinator/src/main/java/org/apache/uniffle/coordinator/strategy/storage/AbstractSelectStorageStrategy.java:
##########
@@ -76,6 +102,72 @@ public void readAndWriteHdfsStorage(FileSystem fs, Path testPath,
}
}
+ @Override
+ public void detectStorage() {
+ uris = Lists.newCopyOnWriteArrayList(remoteStoragePathRankValue.entrySet());
+ if (remoteStoragePathRankValue.size() > 1) {
+ CountDownLatch countDownLatch = new CountDownLatch(uris.size());
+ uris.parallelStream().forEach(uri -> {
+ if (uri.getKey().startsWith(ApplicationManager.getPathSchema().get(0))) {
+ Path remotePath = new Path(uri.getKey());
+ String rssTest = uri.getKey() + "/rssTest-" + getCoordinatorId()
+ + Thread.currentThread().getName();
+ Path testPath = new Path(rssTest);
+ RankValue rankValue = remoteStoragePathRankValue.get(uri.getKey());
+ rankValue.setHealthy(new AtomicBoolean(true));
+ long startWriteTime = System.currentTimeMillis();
+ try {
+ FileSystem fs = HadoopFilesystemProvider.getFilesystem(remotePath, hdfsConf);
+ for (int j = 0; j < readAndWriteTimes; j++) {
+ readAndWriteHdfsStorage(fs, testPath, uri.getKey(), rankValue);
+ }
+ } catch (Exception e) {
+ LOG.error("Storage read and write error, we will not use this remote path {}.", uri, e);
+ rankValue.setHealthy(new AtomicBoolean(false));
+ } finally {
+ sortPathByRankValue(uri.getKey(), rssTest, startWriteTime);
+ }
+ countDownLatch.countDown();
+ }
+ });
+ try {
+ countDownLatch.await();
+ } catch (InterruptedException e) {
+ LOG.error("Failed to detectStorage!");
+ }
+ }
+ }
+
+ @VisibleForTesting
+ public void sortPathByRankValue(
+ String path, String testPath, long startWrite) {
+ RankValue rankValue = remoteStoragePathRankValue.get(path);
+ try {
+ FileSystem fs = HadoopFilesystemProvider.getFilesystem(new Path(path), hdfsConf);
+ fs.delete(new Path(testPath), true);
+ if (rankValue.getHealthy().get()) {
+ rankValue.setCostTime(new AtomicLong(System.currentTimeMillis() - startWrite));
+ }
+ } catch (Exception e) {
+ rankValue.setCostTime(new AtomicLong(Long.MAX_VALUE));
+ LOG.error("Failed to sort, we will not use this remote path {}.", path, e);
+ }
+
+ if (this.getComparator() != null) {
+ uris = Lists.newCopyOnWriteArrayList(
+ remoteStoragePathRankValue.entrySet()).stream().filter(
+ Objects::nonNull).sorted(this.getComparator()).collect(Collectors.toList());
+ } else {
+ uris = Lists.newCopyOnWriteArrayList(
+ remoteStoragePathRankValue.entrySet()).stream().filter(
+ Objects::nonNull).collect(Collectors.toList());
+ }
+ LOG.info("The sorted remote path list is: {}", uris);
+ }
+
+ @Override
+ public abstract Comparator<Map.Entry<String, RankValue>> getComparator();
Review Comment:
Some javadoc please
##########
coordinator/src/main/java/org/apache/uniffle/coordinator/strategy/storage/AbstractSelectStorageStrategy.java:
##########
@@ -18,36 +18,62 @@
package org.apache.uniffle.coordinator.strategy.storage;
import java.io.IOException;
+import java.util.Comparator;
+import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.stream.Collectors;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.Lists;
import org.apache.commons.lang3.RandomUtils;
+import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import org.apache.uniffle.common.exception.RssException;
+import org.apache.uniffle.common.filesystem.HadoopFilesystemProvider;
+import org.apache.uniffle.coordinator.ApplicationManager;
import org.apache.uniffle.coordinator.CoordinatorConf;
import org.apache.uniffle.coordinator.util.CoordinatorUtils;
/**
* This is a simple implementation class, which provides some methods to check whether the path is normal
*/
public abstract class AbstractSelectStorageStrategy implements SelectStorageStrategy {
+
+ private static final Logger LOG = LoggerFactory.getLogger(AbstractSelectStorageStrategy.class);
/**
* store remote path -> application count for assignment strategy
*/
protected final Map<String, RankValue> remoteStoragePathRankValue;
protected final int fileSize;
private final String coordinatorId;
+ private final Configuration hdfsConf;
+ protected List<Map.Entry<String, RankValue>> uris;
+ private int readAndWriteTimes = 1;
public AbstractSelectStorageStrategy(
Map<String, RankValue> remoteStoragePathRankValue,
CoordinatorConf conf) {
this.remoteStoragePathRankValue = remoteStoragePathRankValue;
- fileSize = conf.getInteger(CoordinatorConf.COORDINATOR_REMOTE_STORAGE_SCHEDULE_FILE_SIZE);
+ this.hdfsConf = new Configuration();
+ this.fileSize = conf.getInteger(CoordinatorConf.COORDINATOR_REMOTE_STORAGE_SCHEDULE_FILE_SIZE);
this.coordinatorId = conf.getString(CoordinatorUtils.COORDINATOR_ID, UUID.randomUUID().toString());
+ ApplicationManager.StrategyName strategyName = conf.get(
+ CoordinatorConf.COORDINATOR_REMOTE_STORAGE_SELECT_STRATEGY);
+ if (strategyName == ApplicationManager.StrategyName.IO_SAMPLE) {
Review Comment:
This is weird.....
I believe a better way would be:
1. define a new method: `protected int readAndWriteTimes(Coordinator conf)`
2. override it in the `LowestIOSampleCostSelectStorageStrategy.java`
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] smallzhongfeng commented on a diff in pull request #680: [#678] improvement: Write hdfs files asynchronously when `detectStorage`
Posted by "smallzhongfeng (via GitHub)" <gi...@apache.org>.
smallzhongfeng commented on code in PR #680:
URL: https://github.com/apache/incubator-uniffle/pull/680#discussion_r1125846554
##########
coordinator/src/main/java/org/apache/uniffle/coordinator/strategy/storage/AbstractSelectStorageStrategy.java:
##########
@@ -76,6 +102,72 @@ public void readAndWriteHdfsStorage(FileSystem fs, Path testPath,
}
}
+ @Override
+ public void detectStorage() {
+ uris = Lists.newCopyOnWriteArrayList(remoteStoragePathRankValue.entrySet());
+ if (remoteStoragePathRankValue.size() > 1) {
+ CountDownLatch countDownLatch = new CountDownLatch(uris.size());
+ uris.parallelStream().forEach(uri -> {
+ if (uri.getKey().startsWith(ApplicationManager.getPathSchema().get(0))) {
+ Path remotePath = new Path(uri.getKey());
+ String rssTest = uri.getKey() + "/rssTest-" + getCoordinatorId()
+ + Thread.currentThread().getName();
+ Path testPath = new Path(rssTest);
+ RankValue rankValue = remoteStoragePathRankValue.get(uri.getKey());
+ rankValue.setHealthy(new AtomicBoolean(true));
+ long startWriteTime = System.currentTimeMillis();
+ try {
+ FileSystem fs = HadoopFilesystemProvider.getFilesystem(remotePath, hdfsConf);
+ for (int j = 0; j < readAndWriteTimes; j++) {
+ readAndWriteHdfsStorage(fs, testPath, uri.getKey(), rankValue);
+ }
+ } catch (Exception e) {
+ LOG.error("Storage read and write error, we will not use this remote path {}.", uri, e);
+ rankValue.setHealthy(new AtomicBoolean(false));
+ } finally {
+ sortPathByRankValue(uri.getKey(), rssTest, startWriteTime);
+ }
+ countDownLatch.countDown();
+ }
+ });
+ try {
+ countDownLatch.await();
+ } catch (InterruptedException e) {
+ LOG.error("Failed to detectStorage!");
+ }
+ }
+ }
+
+ @VisibleForTesting
+ public void sortPathByRankValue(
+ String path, String testPath, long startWrite) {
+ RankValue rankValue = remoteStoragePathRankValue.get(path);
+ try {
+ FileSystem fs = HadoopFilesystemProvider.getFilesystem(new Path(path), hdfsConf);
+ fs.delete(new Path(testPath), true);
+ if (rankValue.getHealthy().get()) {
+ rankValue.setCostTime(new AtomicLong(System.currentTimeMillis() - startWrite));
+ }
+ } catch (Exception e) {
+ rankValue.setCostTime(new AtomicLong(Long.MAX_VALUE));
+ LOG.error("Failed to sort, we will not use this remote path {}.", path, e);
+ }
+
+ if (this.getComparator() != null) {
+ uris = Lists.newCopyOnWriteArrayList(
+ remoteStoragePathRankValue.entrySet()).stream().filter(
+ Objects::nonNull).sorted(this.getComparator()).collect(Collectors.toList());
+ } else {
+ uris = Lists.newCopyOnWriteArrayList(
+ remoteStoragePathRankValue.entrySet()).stream().filter(
+ Objects::nonNull).collect(Collectors.toList());
+ }
+ LOG.info("The sorted remote path list is: {}", uris);
+ }
+
+ @Override
+ public abstract Comparator<Map.Entry<String, RankValue>> getComparator();
Review Comment:
Added.
##########
coordinator/src/main/java/org/apache/uniffle/coordinator/strategy/storage/SelectStorageStrategy.java:
##########
@@ -17,11 +17,15 @@
package org.apache.uniffle.coordinator.strategy.storage;
+import java.util.Comparator;
+
import org.apache.uniffle.common.RemoteStorageInfo;
public interface SelectStorageStrategy {
void detectStorage();
RemoteStorageInfo pickStorage(String appId);
+
+ Comparator getComparator();
Review Comment:
Done.
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] codecov-commenter commented on pull request #680: [#678] improvement: Write hdfs files asynchronously when `detectStorage`
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #680:
URL: https://github.com/apache/incubator-uniffle/pull/680#issuecomment-1453204438
# [Codecov](https://codecov.io/gh/apache/incubator-uniffle/pull/680?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#680](https://codecov.io/gh/apache/incubator-uniffle/pull/680?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (c186d7d) into [master](https://codecov.io/gh/apache/incubator-uniffle/commit/1951376f28fe4f91cbe051c6fa8600f52dede8e3?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (1951376) will **increase** coverage by `0.02%`.
> The diff coverage is `80.00%`.
```diff
@@ Coverage Diff @@
## master #680 +/- ##
============================================
+ Coverage 63.05% 63.07% +0.02%
- Complexity 1837 1841 +4
============================================
Files 207 207
Lines 10677 10686 +9
Branches 1067 1065 -2
============================================
+ Hits 6732 6740 +8
- Misses 3598 3601 +3
+ Partials 347 345 -2
```
| [Impacted Files](https://codecov.io/gh/apache/incubator-uniffle/pull/680?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [...orage/LowestIOSampleCostSelectStorageStrategy.java](https://codecov.io/gh/apache/incubator-uniffle/pull/680?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Y29vcmRpbmF0b3Ivc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3VuaWZmbGUvY29vcmRpbmF0b3Ivc3RyYXRlZ3kvc3RvcmFnZS9Mb3dlc3RJT1NhbXBsZUNvc3RTZWxlY3RTdG9yYWdlU3RyYXRlZ3kuamF2YQ==) | `79.66% <77.77%> (+8.75%)` | :arrow_up: |
| [...ategy/storage/AppBalanceSelectStorageStrategy.java](https://codecov.io/gh/apache/incubator-uniffle/pull/680?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Y29vcmRpbmF0b3Ivc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3VuaWZmbGUvY29vcmRpbmF0b3Ivc3RyYXRlZ3kvc3RvcmFnZS9BcHBCYWxhbmNlU2VsZWN0U3RvcmFnZVN0cmF0ZWd5LmphdmE=) | `80.00% <82.35%> (+8.00%)` | :arrow_up: |
| [.../java/org/apache/uniffle/server/ShuffleServer.java](https://codecov.io/gh/apache/incubator-uniffle/pull/680?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2VydmVyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS91bmlmZmxlL3NlcnZlci9TaHVmZmxlU2VydmVyLmphdmE=) | `59.60% <0.00%> (-3.95%)` | :arrow_down: |
:mega: We’re building smart automated test selection to slash your CI/CD build times. [Learn more](https://about.codecov.io/iterative-testing/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] xianjingfeng commented on a diff in pull request #680: [#678] improvement: Write hdfs files asynchronously when `detectStorage`
Posted by "xianjingfeng (via GitHub)" <gi...@apache.org>.
xianjingfeng commented on code in PR #680:
URL: https://github.com/apache/incubator-uniffle/pull/680#discussion_r1125254116
##########
coordinator/src/main/java/org/apache/uniffle/coordinator/strategy/storage/LowestIOSampleCostSelectStorageStrategy.java:
##########
@@ -99,26 +100,38 @@ public void sortPathByRankValue(
public void detectStorage() {
Review Comment:
Could you merge `detectStorage` of `LowestIOSampleCostSelectStorageStrategy` and `AppBalanceSelectStorageStrategy`.
The logic is almost the same.
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] advancedxy merged pull request #680: [#678] improvement: Write hdfs files asynchronously when `detectStorage`
Posted by "advancedxy (via GitHub)" <gi...@apache.org>.
advancedxy merged PR #680:
URL: https://github.com/apache/incubator-uniffle/pull/680
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] xianjingfeng commented on pull request #680: [#678] improvement: Write hdfs files asynchronously when `detectStorage`
Posted by "xianjingfeng (via GitHub)" <gi...@apache.org>.
xianjingfeng commented on PR #680:
URL: https://github.com/apache/incubator-uniffle/pull/680#issuecomment-1454347341
> This doesn't seem to be caused by this pr.
>
> ```
> Errors:
> Error: QuorumTest.case2:337->registerShuffleServer:281 » StatusRuntime DEADLINE_EXCEE...
> ```
Same problem as #667 .
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] xianjingfeng commented on a diff in pull request #680: [#678] improvement: Write hdfs files asynchronously when `detectStorage`
Posted by "xianjingfeng (via GitHub)" <gi...@apache.org>.
xianjingfeng commented on code in PR #680:
URL: https://github.com/apache/incubator-uniffle/pull/680#discussion_r1125587954
##########
coordinator/src/main/java/org/apache/uniffle/coordinator/strategy/storage/AbstractSelectStorageStrategy.java:
##########
@@ -76,6 +92,50 @@ public void readAndWriteHdfsStorage(FileSystem fs, Path testPath,
}
}
+ @Override
+ public void detectStorage() {
+ uris = Lists.newCopyOnWriteArrayList(remoteStoragePathRankValue.entrySet());
+ if (remoteStoragePathRankValue.size() > 1) {
+ CountDownLatch countDownLatch = new CountDownLatch(uris.size());
+ uris.parallelStream().forEach(uri -> {
+ if (uri.getKey().startsWith(ApplicationManager.getPathSchema().get(0))) {
+ Path remotePath = new Path(uri.getKey());
+ String rssTest = uri.getKey() + "/rssTest-" + getCoordinatorId()
+ + Thread.currentThread().getName();
+ Path testPath = new Path(rssTest);
+ RankValue rankValue = remoteStoragePathRankValue.get(uri.getKey());
+ rankValue.setHealthy(new AtomicBoolean(true));
+ long startWriteTime = System.currentTimeMillis();
+ try {
+ FileSystem fs = HadoopFilesystemProvider.getFilesystem(remotePath, hdfsConf);
+ for (int j = 0; j < readAndWriteTimes; j++) {
Review Comment:
Do you have any suggestions here? @jerqi @advancedxy
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] xianjingfeng commented on a diff in pull request #680: [#678] improvement: Write hdfs files asynchronously when `detectStorage`
Posted by "xianjingfeng (via GitHub)" <gi...@apache.org>.
xianjingfeng commented on code in PR #680:
URL: https://github.com/apache/incubator-uniffle/pull/680#discussion_r1125251152
##########
coordinator/src/main/java/org/apache/uniffle/coordinator/strategy/storage/AppBalanceSelectStorageStrategy.java:
##########
@@ -84,23 +85,35 @@ public void sortPathByRankValue(String path, String test) {
public void detectStorage() {
uris = Lists.newCopyOnWriteArrayList(remoteStoragePathRankValue.entrySet());
if (remoteStoragePathRankValue.size() > 1) {
+ CountDownLatch countDownLatch = new CountDownLatch(uris.size());
for (Map.Entry<String, RankValue> uri : uris) {
- if (uri.getKey().startsWith(ApplicationManager.getPathSchema().get(0))) {
- RankValue rankValue = remoteStoragePathRankValue.get(uri.getKey());
- rankValue.setHealthy(new AtomicBoolean(true));
- Path remotePath = new Path(uri.getKey());
- String rssTest = uri.getKey() + "/rssTest-" + getCoordinatorId();
- Path testPath = new Path(rssTest);
- try {
- FileSystem fs = HadoopFilesystemProvider.getFilesystem(remotePath, hdfsConf);
- readAndWriteHdfsStorage(fs, testPath, uri.getKey(), rankValue);
- } catch (Exception e) {
- rankValue.setHealthy(new AtomicBoolean(false));
- LOG.error("Storage read and write error, we will not use this remote path {}.", uri, e);
- } finally {
- sortPathByRankValue(uri.getKey(), rssTest);
+ Thread detectThread = new Thread(() -> {
Review Comment:
I mean use `uris.stream().parallel()`
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] smallzhongfeng commented on a diff in pull request #680: [#678] improvement: Write hdfs files asynchronously when `detectStorage`
Posted by "smallzhongfeng (via GitHub)" <gi...@apache.org>.
smallzhongfeng commented on code in PR #680:
URL: https://github.com/apache/incubator-uniffle/pull/680#discussion_r1125453742
##########
coordinator/src/main/java/org/apache/uniffle/coordinator/strategy/storage/AbstractSelectStorageStrategy.java:
##########
@@ -76,6 +92,50 @@ public void readAndWriteHdfsStorage(FileSystem fs, Path testPath,
}
}
+ @Override
+ public void detectStorage() {
+ uris = Lists.newCopyOnWriteArrayList(remoteStoragePathRankValue.entrySet());
+ if (remoteStoragePathRankValue.size() > 1) {
+ CountDownLatch countDownLatch = new CountDownLatch(uris.size());
+ uris.parallelStream().forEach(uri -> {
+ if (uri.getKey().startsWith(ApplicationManager.getPathSchema().get(0))) {
+ Path remotePath = new Path(uri.getKey());
+ String rssTest = uri.getKey() + "/rssTest-" + getCoordinatorId()
+ + Thread.currentThread().getName();
+ Path testPath = new Path(rssTest);
+ RankValue rankValue = remoteStoragePathRankValue.get(uri.getKey());
+ rankValue.setHealthy(new AtomicBoolean(true));
+ long startWriteTime = System.currentTimeMillis();
+ try {
+ FileSystem fs = HadoopFilesystemProvider.getFilesystem(remotePath, hdfsConf);
+ for (int j = 0; j < readAndWriteTimes; j++) {
+ readAndWriteHdfsStorage(fs, testPath, uri.getKey(), rankValue);
+ }
+ } catch (Exception e) {
+ LOG.error("Storage read and write error, we will not use this remote path {}.", uri, e);
+ rankValue.setHealthy(new AtomicBoolean(false));
+ } finally {
+ if (this instanceof LowestIOSampleCostSelectStorageStrategy) {
+ ((LowestIOSampleCostSelectStorageStrategy) this)
+ .sortPathByRankValue(uri.getKey(), rssTest, startWriteTime, hdfsConf);
+ } else if (this instanceof AppBalanceSelectStorageStrategy) {
+ ((AppBalanceSelectStorageStrategy) this)
+ .sortPathByRankValue(uri.getKey(), rssTest, hdfsConf);
Review Comment:
I don't quite understand the meaning of the second point, why the getComparator of AppBalanceSelectStorageStrategy returns null? @xianjingfeng
##########
coordinator/src/main/java/org/apache/uniffle/coordinator/strategy/storage/AbstractSelectStorageStrategy.java:
##########
@@ -76,6 +92,50 @@ public void readAndWriteHdfsStorage(FileSystem fs, Path testPath,
}
}
+ @Override
+ public void detectStorage() {
+ uris = Lists.newCopyOnWriteArrayList(remoteStoragePathRankValue.entrySet());
+ if (remoteStoragePathRankValue.size() > 1) {
+ CountDownLatch countDownLatch = new CountDownLatch(uris.size());
+ uris.parallelStream().forEach(uri -> {
+ if (uri.getKey().startsWith(ApplicationManager.getPathSchema().get(0))) {
+ Path remotePath = new Path(uri.getKey());
+ String rssTest = uri.getKey() + "/rssTest-" + getCoordinatorId()
+ + Thread.currentThread().getName();
+ Path testPath = new Path(rssTest);
+ RankValue rankValue = remoteStoragePathRankValue.get(uri.getKey());
+ rankValue.setHealthy(new AtomicBoolean(true));
+ long startWriteTime = System.currentTimeMillis();
+ try {
+ FileSystem fs = HadoopFilesystemProvider.getFilesystem(remotePath, hdfsConf);
+ for (int j = 0; j < readAndWriteTimes; j++) {
+ readAndWriteHdfsStorage(fs, testPath, uri.getKey(), rankValue);
+ }
+ } catch (Exception e) {
+ LOG.error("Storage read and write error, we will not use this remote path {}.", uri, e);
+ rankValue.setHealthy(new AtomicBoolean(false));
+ } finally {
+ if (this instanceof LowestIOSampleCostSelectStorageStrategy) {
+ ((LowestIOSampleCostSelectStorageStrategy) this)
+ .sortPathByRankValue(uri.getKey(), rssTest, startWriteTime, hdfsConf);
+ } else if (this instanceof AppBalanceSelectStorageStrategy) {
+ ((AppBalanceSelectStorageStrategy) this)
+ .sortPathByRankValue(uri.getKey(), rssTest, hdfsConf);
Review Comment:
I don't quite understand the meaning of the second point, why the getComparator of AppBalanceSelectStorageStrategy returns null? @xianjingfeng
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] smallzhongfeng commented on a diff in pull request #680: [#678] improvement: Write hdfs files asynchronously when `detectStorage`
Posted by "smallzhongfeng (via GitHub)" <gi...@apache.org>.
smallzhongfeng commented on code in PR #680:
URL: https://github.com/apache/incubator-uniffle/pull/680#discussion_r1125453840
##########
coordinator/src/main/java/org/apache/uniffle/coordinator/strategy/storage/AbstractSelectStorageStrategy.java:
##########
@@ -76,6 +92,50 @@ public void readAndWriteHdfsStorage(FileSystem fs, Path testPath,
}
}
+ @Override
+ public void detectStorage() {
+ uris = Lists.newCopyOnWriteArrayList(remoteStoragePathRankValue.entrySet());
+ if (remoteStoragePathRankValue.size() > 1) {
+ CountDownLatch countDownLatch = new CountDownLatch(uris.size());
+ uris.parallelStream().forEach(uri -> {
+ if (uri.getKey().startsWith(ApplicationManager.getPathSchema().get(0))) {
+ Path remotePath = new Path(uri.getKey());
+ String rssTest = uri.getKey() + "/rssTest-" + getCoordinatorId()
+ + Thread.currentThread().getName();
+ Path testPath = new Path(rssTest);
+ RankValue rankValue = remoteStoragePathRankValue.get(uri.getKey());
+ rankValue.setHealthy(new AtomicBoolean(true));
+ long startWriteTime = System.currentTimeMillis();
+ try {
+ FileSystem fs = HadoopFilesystemProvider.getFilesystem(remotePath, hdfsConf);
+ for (int j = 0; j < readAndWriteTimes; j++) {
+ readAndWriteHdfsStorage(fs, testPath, uri.getKey(), rankValue);
+ }
+ } catch (Exception e) {
+ LOG.error("Storage read and write error, we will not use this remote path {}.", uri, e);
+ rankValue.setHealthy(new AtomicBoolean(false));
+ } finally {
+ if (this instanceof LowestIOSampleCostSelectStorageStrategy) {
+ ((LowestIOSampleCostSelectStorageStrategy) this)
+ .sortPathByRankValue(uri.getKey(), rssTest, startWriteTime, hdfsConf);
+ } else if (this instanceof AppBalanceSelectStorageStrategy) {
+ ((AppBalanceSelectStorageStrategy) this)
+ .sortPathByRankValue(uri.getKey(), rssTest, hdfsConf);
Review Comment:
It's different when
```rankValue.setCostTime(new AtomicLong(System.currentTimeMillis() - startWrite));```
and sort `uris`.
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] smallzhongfeng commented on a diff in pull request #680: [#678] improvement: Write hdfs files asynchronously when `detectStorage`
Posted by "smallzhongfeng (via GitHub)" <gi...@apache.org>.
smallzhongfeng commented on code in PR #680:
URL: https://github.com/apache/incubator-uniffle/pull/680#discussion_r1126076118
##########
coordinator/src/test/java/org/apache/uniffle/coordinator/strategy/storage/LowestIOSampleCostSelectStorageStrategyTest.java:
##########
@@ -90,6 +91,7 @@ public void selectStorageTest() throws Exception {
assertEquals(0, applicationManager.getRemoteStoragePathRankValue().get(remoteStorage2).getAppNum().get());
// compare with two remote path
+ final Configuration configuration = new Configuration();
Review Comment:
Updated.
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] xianjingfeng commented on a diff in pull request #680: [#678] improvement: Write hdfs files asynchronously when `detectStorage`
Posted by "xianjingfeng (via GitHub)" <gi...@apache.org>.
xianjingfeng commented on code in PR #680:
URL: https://github.com/apache/incubator-uniffle/pull/680#discussion_r1125424513
##########
coordinator/src/main/java/org/apache/uniffle/coordinator/strategy/storage/AbstractSelectStorageStrategy.java:
##########
@@ -76,6 +92,50 @@ public void readAndWriteHdfsStorage(FileSystem fs, Path testPath,
}
}
+ @Override
+ public void detectStorage() {
+ uris = Lists.newCopyOnWriteArrayList(remoteStoragePathRankValue.entrySet());
+ if (remoteStoragePathRankValue.size() > 1) {
+ CountDownLatch countDownLatch = new CountDownLatch(uris.size());
+ uris.parallelStream().forEach(uri -> {
+ if (uri.getKey().startsWith(ApplicationManager.getPathSchema().get(0))) {
+ Path remotePath = new Path(uri.getKey());
+ String rssTest = uri.getKey() + "/rssTest-" + getCoordinatorId()
+ + Thread.currentThread().getName();
+ Path testPath = new Path(rssTest);
+ RankValue rankValue = remoteStoragePathRankValue.get(uri.getKey());
+ rankValue.setHealthy(new AtomicBoolean(true));
+ long startWriteTime = System.currentTimeMillis();
+ try {
+ FileSystem fs = HadoopFilesystemProvider.getFilesystem(remotePath, hdfsConf);
+ for (int j = 0; j < readAndWriteTimes; j++) {
+ readAndWriteHdfsStorage(fs, testPath, uri.getKey(), rankValue);
+ }
+ } catch (Exception e) {
+ LOG.error("Storage read and write error, we will not use this remote path {}.", uri, e);
+ rankValue.setHealthy(new AtomicBoolean(false));
+ } finally {
+ if (this instanceof LowestIOSampleCostSelectStorageStrategy) {
+ ((LowestIOSampleCostSelectStorageStrategy) this)
+ .sortPathByRankValue(uri.getKey(), rssTest, startWriteTime, hdfsConf);
+ } else if (this instanceof AppBalanceSelectStorageStrategy) {
+ ((AppBalanceSelectStorageStrategy) this)
+ .sortPathByRankValue(uri.getKey(), rssTest, hdfsConf);
+ } else {
+ LOG.error("Failed to sort path by detectStorage!");
+ }
+ countDownLatch.countDown();
Review Comment:
It is better to move this line inside the `finally` block.
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] xianjingfeng commented on a diff in pull request #680: [#678] improvement: Write hdfs files asynchronously when `detectStorage`
Posted by "xianjingfeng (via GitHub)" <gi...@apache.org>.
xianjingfeng commented on code in PR #680:
URL: https://github.com/apache/incubator-uniffle/pull/680#discussion_r1125461553
##########
coordinator/src/main/java/org/apache/uniffle/coordinator/strategy/storage/AbstractSelectStorageStrategy.java:
##########
@@ -76,6 +92,50 @@ public void readAndWriteHdfsStorage(FileSystem fs, Path testPath,
}
}
+ @Override
+ public void detectStorage() {
+ uris = Lists.newCopyOnWriteArrayList(remoteStoragePathRankValue.entrySet());
+ if (remoteStoragePathRankValue.size() > 1) {
+ CountDownLatch countDownLatch = new CountDownLatch(uris.size());
+ uris.parallelStream().forEach(uri -> {
+ if (uri.getKey().startsWith(ApplicationManager.getPathSchema().get(0))) {
+ Path remotePath = new Path(uri.getKey());
+ String rssTest = uri.getKey() + "/rssTest-" + getCoordinatorId()
+ + Thread.currentThread().getName();
+ Path testPath = new Path(rssTest);
+ RankValue rankValue = remoteStoragePathRankValue.get(uri.getKey());
+ rankValue.setHealthy(new AtomicBoolean(true));
+ long startWriteTime = System.currentTimeMillis();
+ try {
+ FileSystem fs = HadoopFilesystemProvider.getFilesystem(remotePath, hdfsConf);
+ for (int j = 0; j < readAndWriteTimes; j++) {
+ readAndWriteHdfsStorage(fs, testPath, uri.getKey(), rankValue);
+ }
+ } catch (Exception e) {
+ LOG.error("Storage read and write error, we will not use this remote path {}.", uri, e);
+ rankValue.setHealthy(new AtomicBoolean(false));
+ } finally {
+ if (this instanceof LowestIOSampleCostSelectStorageStrategy) {
+ ((LowestIOSampleCostSelectStorageStrategy) this)
+ .sortPathByRankValue(uri.getKey(), rssTest, startWriteTime, hdfsConf);
+ } else if (this instanceof AppBalanceSelectStorageStrategy) {
+ ((AppBalanceSelectStorageStrategy) this)
+ .sortPathByRankValue(uri.getKey(), rssTest, hdfsConf);
Review Comment:
`rankValue` is useless in `AppBalanceSelectStorageStrategy`. The difference between `sortPathByRankValue` of the two strategies is that one is sorted and the other is not.
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] xianjingfeng commented on pull request #680: [#678] improvement: Write hdfs files asynchronously when `detectStorage`
Posted by "xianjingfeng (via GitHub)" <gi...@apache.org>.
xianjingfeng commented on PR #680:
URL: https://github.com/apache/incubator-uniffle/pull/680#issuecomment-1453577392
Maybe we should use CountdownLaunch + (parallelStream or thread pool).
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] xianjingfeng commented on a diff in pull request #680: [#678] improvement: Write hdfs files asynchronously when `detectStorage`
Posted by "xianjingfeng (via GitHub)" <gi...@apache.org>.
xianjingfeng commented on code in PR #680:
URL: https://github.com/apache/incubator-uniffle/pull/680#discussion_r1125587410
##########
coordinator/src/main/java/org/apache/uniffle/coordinator/strategy/storage/AbstractSelectStorageStrategy.java:
##########
@@ -76,6 +92,50 @@ public void readAndWriteHdfsStorage(FileSystem fs, Path testPath,
}
}
+ @Override
+ public void detectStorage() {
+ uris = Lists.newCopyOnWriteArrayList(remoteStoragePathRankValue.entrySet());
+ if (remoteStoragePathRankValue.size() > 1) {
+ CountDownLatch countDownLatch = new CountDownLatch(uris.size());
+ uris.parallelStream().forEach(uri -> {
+ if (uri.getKey().startsWith(ApplicationManager.getPathSchema().get(0))) {
+ Path remotePath = new Path(uri.getKey());
+ String rssTest = uri.getKey() + "/rssTest-" + getCoordinatorId()
+ + Thread.currentThread().getName();
+ Path testPath = new Path(rssTest);
+ RankValue rankValue = remoteStoragePathRankValue.get(uri.getKey());
+ rankValue.setHealthy(new AtomicBoolean(true));
+ long startWriteTime = System.currentTimeMillis();
+ try {
+ FileSystem fs = HadoopFilesystemProvider.getFilesystem(remotePath, hdfsConf);
+ for (int j = 0; j < readAndWriteTimes; j++) {
+ readAndWriteHdfsStorage(fs, testPath, uri.getKey(), rankValue);
+ }
+ } catch (Exception e) {
+ LOG.error("Storage read and write error, we will not use this remote path {}.", uri, e);
+ rankValue.setHealthy(new AtomicBoolean(false));
+ } finally {
+ if (this instanceof LowestIOSampleCostSelectStorageStrategy) {
+ ((LowestIOSampleCostSelectStorageStrategy) this)
+ .sortPathByRankValue(uri.getKey(), rssTest, startWriteTime, hdfsConf);
+ } else if (this instanceof AppBalanceSelectStorageStrategy) {
+ ((AppBalanceSelectStorageStrategy) this)
+ .sortPathByRankValue(uri.getKey(), rssTest, hdfsConf);
Review Comment:
Yes
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] xianjingfeng commented on a diff in pull request #680: [#678] improvement: Write hdfs files asynchronously when `detectStorage`
Posted by "xianjingfeng (via GitHub)" <gi...@apache.org>.
xianjingfeng commented on code in PR #680:
URL: https://github.com/apache/incubator-uniffle/pull/680#discussion_r1125587631
##########
coordinator/src/main/java/org/apache/uniffle/coordinator/strategy/storage/AppBalanceSelectStorageStrategy.java:
##########
@@ -132,4 +76,9 @@ public synchronized RemoteStorageInfo pickStorage(String appId) {
LOG.warn("No remote storage is available, we will default to the first.");
return availableRemoteStorageInfo.values().iterator().next();
}
+
+ @Override
+ public Comparator<Map.Entry<String, RankValue>> getComparator() {
+ return (o1, o2) -> 0;
Review Comment:
I think return null is better, We don't need to sort the paths in this case.
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] smallzhongfeng commented on a diff in pull request #680: [#678] improvement: Write hdfs files asynchronously when `detectStorage`
Posted by "smallzhongfeng (via GitHub)" <gi...@apache.org>.
smallzhongfeng commented on code in PR #680:
URL: https://github.com/apache/incubator-uniffle/pull/680#discussion_r1125596476
##########
coordinator/src/main/java/org/apache/uniffle/coordinator/strategy/storage/AppBalanceSelectStorageStrategy.java:
##########
@@ -132,4 +76,9 @@ public synchronized RemoteStorageInfo pickStorage(String appId) {
LOG.warn("No remote storage is available, we will default to the first.");
return availableRemoteStorageInfo.values().iterator().next();
}
+
+ @Override
+ public Comparator<Map.Entry<String, RankValue>> getComparator() {
+ return (o1, o2) -> 0;
Review Comment:
Ok, updated.
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] smallzhongfeng commented on a diff in pull request #680: [#678] improvement: Write hdfs files asynchronously when `detectStorage`
Posted by "smallzhongfeng (via GitHub)" <gi...@apache.org>.
smallzhongfeng commented on code in PR #680:
URL: https://github.com/apache/incubator-uniffle/pull/680#discussion_r1125847240
##########
coordinator/src/main/java/org/apache/uniffle/coordinator/strategy/storage/AbstractSelectStorageStrategy.java:
##########
@@ -18,36 +18,62 @@
package org.apache.uniffle.coordinator.strategy.storage;
import java.io.IOException;
+import java.util.Comparator;
+import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.stream.Collectors;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.Lists;
import org.apache.commons.lang3.RandomUtils;
+import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import org.apache.uniffle.common.exception.RssException;
+import org.apache.uniffle.common.filesystem.HadoopFilesystemProvider;
+import org.apache.uniffle.coordinator.ApplicationManager;
import org.apache.uniffle.coordinator.CoordinatorConf;
import org.apache.uniffle.coordinator.util.CoordinatorUtils;
/**
* This is a simple implementation class, which provides some methods to check whether the path is normal
*/
public abstract class AbstractSelectStorageStrategy implements SelectStorageStrategy {
+
+ private static final Logger LOG = LoggerFactory.getLogger(AbstractSelectStorageStrategy.class);
/**
* store remote path -> application count for assignment strategy
*/
protected final Map<String, RankValue> remoteStoragePathRankValue;
protected final int fileSize;
private final String coordinatorId;
+ private final Configuration hdfsConf;
+ protected List<Map.Entry<String, RankValue>> uris;
+ private int readAndWriteTimes = 1;
public AbstractSelectStorageStrategy(
Map<String, RankValue> remoteStoragePathRankValue,
CoordinatorConf conf) {
this.remoteStoragePathRankValue = remoteStoragePathRankValue;
- fileSize = conf.getInteger(CoordinatorConf.COORDINATOR_REMOTE_STORAGE_SCHEDULE_FILE_SIZE);
+ this.hdfsConf = new Configuration();
+ this.fileSize = conf.getInteger(CoordinatorConf.COORDINATOR_REMOTE_STORAGE_SCHEDULE_FILE_SIZE);
this.coordinatorId = conf.getString(CoordinatorUtils.COORDINATOR_ID, UUID.randomUUID().toString());
+ ApplicationManager.StrategyName strategyName = conf.get(
+ CoordinatorConf.COORDINATOR_REMOTE_STORAGE_SELECT_STRATEGY);
+ if (strategyName == ApplicationManager.StrategyName.IO_SAMPLE) {
Review Comment:
Good idea! Updated.
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] xianjingfeng commented on a diff in pull request #680: [#678] improvement: Write hdfs files asynchronously when `detectStorage`
Posted by "xianjingfeng (via GitHub)" <gi...@apache.org>.
xianjingfeng commented on code in PR #680:
URL: https://github.com/apache/incubator-uniffle/pull/680#discussion_r1125423521
##########
coordinator/src/main/java/org/apache/uniffle/coordinator/strategy/storage/AbstractSelectStorageStrategy.java:
##########
@@ -76,6 +92,50 @@ public void readAndWriteHdfsStorage(FileSystem fs, Path testPath,
}
}
+ @Override
+ public void detectStorage() {
+ uris = Lists.newCopyOnWriteArrayList(remoteStoragePathRankValue.entrySet());
+ if (remoteStoragePathRankValue.size() > 1) {
+ CountDownLatch countDownLatch = new CountDownLatch(uris.size());
+ uris.parallelStream().forEach(uri -> {
+ if (uri.getKey().startsWith(ApplicationManager.getPathSchema().get(0))) {
+ Path remotePath = new Path(uri.getKey());
+ String rssTest = uri.getKey() + "/rssTest-" + getCoordinatorId()
+ + Thread.currentThread().getName();
+ Path testPath = new Path(rssTest);
+ RankValue rankValue = remoteStoragePathRankValue.get(uri.getKey());
+ rankValue.setHealthy(new AtomicBoolean(true));
+ long startWriteTime = System.currentTimeMillis();
+ try {
+ FileSystem fs = HadoopFilesystemProvider.getFilesystem(remotePath, hdfsConf);
+ for (int j = 0; j < readAndWriteTimes; j++) {
+ readAndWriteHdfsStorage(fs, testPath, uri.getKey(), rankValue);
+ }
+ } catch (Exception e) {
+ LOG.error("Storage read and write error, we will not use this remote path {}.", uri, e);
+ rankValue.setHealthy(new AtomicBoolean(false));
+ } finally {
+ if (this instanceof LowestIOSampleCostSelectStorageStrategy) {
+ ((LowestIOSampleCostSelectStorageStrategy) this)
+ .sortPathByRankValue(uri.getKey(), rssTest, startWriteTime, hdfsConf);
+ } else if (this instanceof AppBalanceSelectStorageStrategy) {
+ ((AppBalanceSelectStorageStrategy) this)
+ .sortPathByRankValue(uri.getKey(), rssTest, hdfsConf);
Review Comment:
I think `sortPathByRankValue` can be merged too.
1. We can add a abstract method named `getComparator` to SelectStorageStrategy and move `sortPathByRankValue` to `AbstractSelectStorageStrategy`.
2. Return null in the overrided method in `AppBalanceSelectStorageStrategy` and return a comparator in `LowestIOSampleCostSelectStorageStrategy`
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] xianjingfeng commented on a diff in pull request #680: [#678] improvement: Write hdfs files asynchronously when `detectStorage`
Posted by "xianjingfeng (via GitHub)" <gi...@apache.org>.
xianjingfeng commented on code in PR #680:
URL: https://github.com/apache/incubator-uniffle/pull/680#discussion_r1125424602
##########
coordinator/src/main/java/org/apache/uniffle/coordinator/strategy/storage/AppBalanceSelectStorageStrategy.java:
##########
@@ -84,23 +85,35 @@ public void sortPathByRankValue(String path, String test) {
public void detectStorage() {
uris = Lists.newCopyOnWriteArrayList(remoteStoragePathRankValue.entrySet());
if (remoteStoragePathRankValue.size() > 1) {
+ CountDownLatch countDownLatch = new CountDownLatch(uris.size());
for (Map.Entry<String, RankValue> uri : uris) {
- if (uri.getKey().startsWith(ApplicationManager.getPathSchema().get(0))) {
- RankValue rankValue = remoteStoragePathRankValue.get(uri.getKey());
- rankValue.setHealthy(new AtomicBoolean(true));
- Path remotePath = new Path(uri.getKey());
- String rssTest = uri.getKey() + "/rssTest-" + getCoordinatorId();
- Path testPath = new Path(rssTest);
- try {
- FileSystem fs = HadoopFilesystemProvider.getFilesystem(remotePath, hdfsConf);
- readAndWriteHdfsStorage(fs, testPath, uri.getKey(), rankValue);
- } catch (Exception e) {
- rankValue.setHealthy(new AtomicBoolean(false));
- LOG.error("Storage read and write error, we will not use this remote path {}.", uri, e);
- } finally {
- sortPathByRankValue(uri.getKey(), rssTest);
+ Thread detectThread = new Thread(() -> {
Review Comment:
Yes
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] smallzhongfeng commented on a diff in pull request #680: [#678] improvement: Write hdfs files asynchronously when `detectStorage`
Posted by "smallzhongfeng (via GitHub)" <gi...@apache.org>.
smallzhongfeng commented on code in PR #680:
URL: https://github.com/apache/incubator-uniffle/pull/680#discussion_r1125453840
##########
coordinator/src/main/java/org/apache/uniffle/coordinator/strategy/storage/AbstractSelectStorageStrategy.java:
##########
@@ -76,6 +92,50 @@ public void readAndWriteHdfsStorage(FileSystem fs, Path testPath,
}
}
+ @Override
+ public void detectStorage() {
+ uris = Lists.newCopyOnWriteArrayList(remoteStoragePathRankValue.entrySet());
+ if (remoteStoragePathRankValue.size() > 1) {
+ CountDownLatch countDownLatch = new CountDownLatch(uris.size());
+ uris.parallelStream().forEach(uri -> {
+ if (uri.getKey().startsWith(ApplicationManager.getPathSchema().get(0))) {
+ Path remotePath = new Path(uri.getKey());
+ String rssTest = uri.getKey() + "/rssTest-" + getCoordinatorId()
+ + Thread.currentThread().getName();
+ Path testPath = new Path(rssTest);
+ RankValue rankValue = remoteStoragePathRankValue.get(uri.getKey());
+ rankValue.setHealthy(new AtomicBoolean(true));
+ long startWriteTime = System.currentTimeMillis();
+ try {
+ FileSystem fs = HadoopFilesystemProvider.getFilesystem(remotePath, hdfsConf);
+ for (int j = 0; j < readAndWriteTimes; j++) {
+ readAndWriteHdfsStorage(fs, testPath, uri.getKey(), rankValue);
+ }
+ } catch (Exception e) {
+ LOG.error("Storage read and write error, we will not use this remote path {}.", uri, e);
+ rankValue.setHealthy(new AtomicBoolean(false));
+ } finally {
+ if (this instanceof LowestIOSampleCostSelectStorageStrategy) {
+ ((LowestIOSampleCostSelectStorageStrategy) this)
+ .sortPathByRankValue(uri.getKey(), rssTest, startWriteTime, hdfsConf);
+ } else if (this instanceof AppBalanceSelectStorageStrategy) {
+ ((AppBalanceSelectStorageStrategy) this)
+ .sortPathByRankValue(uri.getKey(), rssTest, hdfsConf);
Review Comment:
It's different when
```rankValue.setCostTime(new AtomicLong(System.currentTimeMillis() - startWrite));``` in IO_SAMPLE
```rankValue.setCostTime(new AtomicLong(0));``` in APP
and sort `uris`.
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org