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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
