From 08424ce408b5e1ee679d15e46ea5b08979511fae Mon Sep 17 00:00:00 2001
From: yantangzhai <tyz0303@163.com>
Date: Wed, 2 Jul 2014 14:55:39 +0800
Subject: [PATCH] [SPARK-2325] Utils.getLocalDir had better check the directory
 and choose a good one instead of choosing the first one directly

---
 .../org/apache/spark/util/DiskChecker.scala   | 46 +++++++++++++++++++
 .../scala/org/apache/spark/util/Utils.scala   |  9 +++-
 2 files changed, 54 insertions(+), 1 deletion(-)
 create mode 100644 core/src/main/scala/org/apache/spark/util/DiskChecker.scala

diff --git a/core/src/main/scala/org/apache/spark/util/DiskChecker.scala b/core/src/main/scala/org/apache/spark/util/DiskChecker.scala
new file mode 100644
index 000000000000..2bdcdae3d220
--- /dev/null
+++ b/core/src/main/scala/org/apache/spark/util/DiskChecker.scala
@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.util
+
+import java.io._
+
+import org.apache.spark.Logging
+
+/**
+ * Various utility methods used by Spark for checking disks.
+ */
+private[spark] object DiskChecker extends Logging {
+
+  def mkdirsWithExistsCheck(dir: File): Boolean = {
+    if (dir.mkdir() || dir.exists()) {
+      return true
+    }
+    var canonDir: File = null
+    try {
+      canonDir = dir.getCanonicalFile()
+    } catch {
+      case ie: IOException => return false
+    }
+    val parent = canonDir.getParent()
+    (parent != null) && (mkdirsWithExistsCheck(new File(parent)) && (canonDir.mkdir() || canonDir.exists()))
+  }
+
+  def checkDir(dir: File): Boolean = {
+    mkdirsWithExistsCheck(dir) && dir.isDirectory() && dir.canRead() && dir.canWrite()
+  }
+}
diff --git a/core/src/main/scala/org/apache/spark/util/Utils.scala b/core/src/main/scala/org/apache/spark/util/Utils.scala
index a2454e120a8a..61651302d570 100644
--- a/core/src/main/scala/org/apache/spark/util/Utils.scala
+++ b/core/src/main/scala/org/apache/spark/util/Utils.scala
@@ -430,7 +430,14 @@ private[spark] object Utils extends Logging {
    * multiple paths.
    */
   def getLocalDir(conf: SparkConf): String = {
-    conf.get("spark.local.dir",  System.getProperty("java.io.tmpdir")).split(',')(0)
+    val localDirs = conf.get("spark.local.dir",  System.getProperty("java.io.tmpdir")).split(',')
+    for (localDir <- localDirs) {
+      if (DiskChecker.checkDir(new File(localDir))) {
+        return localDir
+      }
+    }
+    logWarning("No local dir is good by checked, the first one will be choosed.")
+    localDirs(0)
   }
 
   /**