[SPARK-5931] Changed default behavior of time string conversion.

apache · ilganeli · Mar 27, 2015 · Mar 27, 2015 · Mar 27, 2015 · Mar 27, 2015
commit 5232a36f660022f87b3a8c68a8aa4b514d57fbb1
diff --git a/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala b/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala
@@ -78,15 +78,15 @@ private[spark] class ExecutorAllocationManager(
     Integer.MAX_VALUE)
 
   // How long there must be backlogged tasks for before an addition is triggered (seconds)
-  private val schedulerBacklogTimeoutS = Utils.timeStringToS(conf.get(
+  private val schedulerBacklogTimeoutS = Utils.timeStringAsS(conf.get(
     "spark.dynamicAllocation.schedulerBacklogTimeoutS", "5s"))
 
   // Same as above, but used only after `schedulerBacklogTimeoutS` is exceeded
-  private val sustainedSchedulerBacklogTimeoutS = Utils.timeStringToS(conf.get(
+  private val sustainedSchedulerBacklogTimeoutS = Utils.timeStringAsS(conf.get(
     "spark.dynamicAllocation.sustainedSchedulerBacklogTimeoutS", s"${schedulerBacklogTimeoutS}s"))
 
   // How long an executor must be idle for before it is removed (seconds)
-  private val executorIdleTimeoutS = Utils.timeStringToS(conf.get(
+  private val executorIdleTimeoutS = Utils.timeStringAsS(conf.get(
     "spark.dynamicAllocation.executorIdleTimeoutS", "600s"))
 
   // During testing, the methods to actually kill and add executors are mocked out

diff --git a/core/src/main/scala/org/apache/spark/HeartbeatReceiver.scala b/core/src/main/scala/org/apache/spark/HeartbeatReceiver.scala
@@ -50,12 +50,12 @@ private[spark] class HeartbeatReceiver(sc: SparkContext, scheduler: TaskSchedule
   // executor ID -> timestamp of when the last heartbeat from this executor was received
   private val executorLastSeen = new mutable.HashMap[String, Long]
 
-  private val executorTimeoutMs = Utils.timeStringToMs(sc.conf.get("spark.network.timeout",
+  private val executorTimeoutMs = Utils.timeStringAsMs(sc.conf.get("spark.network.timeout",
       sc.conf.get("spark.storage.blockManagerSlaveTimeout", "120s")))
 
   // "spark.network.timeoutInterval" uses "seconds", while
   // "spark.storage.blockManagerTimeoutIntervalMs" uses "milliseconds"
-  private val checkTimeoutIntervalMs = Utils.timeStringToMs(
+  private val checkTimeoutIntervalMs = Utils.timeStringAsMs(
     sc.conf.get("spark.network.timeoutInterval",
       sc.conf.get("spark.storage.blockManagerTimeoutIntervalMs", "60s")))
 

diff --git a/core/src/main/scala/org/apache/spark/executor/Executor.scala b/core/src/main/scala/org/apache/spark/executor/Executor.scala
@@ -442,7 +442,7 @@ private[spark] class Executor(
    * This thread stops running when the executor is stopped.
    */
   private def startDriverHeartbeater(): Unit = {
-    val intervalMs = Utils.timeStringToMs(conf.get("spark.executor.heartbeatInterval", "10s"))
+    val intervalMs = Utils.timeStringAsMs(conf.get("spark.executor.heartbeatInterval", "10s"))
     val thread = new Thread() {
       override def run() {
         // Sleep a random intervalMs so the heartbeats don't end up in sync

diff --git a/core/src/main/scala/org/apache/spark/network/nio/ConnectionManager.scala b/core/src/main/scala/org/apache/spark/network/nio/ConnectionManager.scala
@@ -82,7 +82,7 @@ private[nio] class ConnectionManager(
     new HashedWheelTimer(Utils.namedThreadFactory("AckTimeoutMonitor"))
 
   private val ackTimeout =
-    Utils.timeStringToS(conf.get("spark.core.connection.ack.wait.timeout",
+    Utils.timeStringAsS(conf.get("spark.core.connection.ack.wait.timeout",
       conf.get("spark.network.timeout", "120s")))
 
   // Get the thread counts from the Spark Configuration.

diff --git a/core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala b/core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala
@@ -63,10 +63,10 @@ private[spark] class TaskSchedulerImpl(
 
   // How often to check for speculative tasks
   val SPECULATION_INTERVAL_MS =
-    Utils.timeStringToMs(conf.get("spark.speculation.interval", "100ms"))
+    Utils.timeStringAsMs(conf.get("spark.speculation.interval", "100ms"))
 
   // Threshold above which we warn user initial TaskSet may be starved
-  val STARVATION_TIMEOUT_MS = Utils.timeStringToMs(conf.get("spark.starvation.timeout", "15000ms"))
+  val STARVATION_TIMEOUT_MS = Utils.timeStringAsMs(conf.get("spark.starvation.timeout", "15000ms"))
 
   // CPUs to request per task
   val CPUS_PER_TASK = conf.getInt("spark.task.cpus", 1)

diff --git a/core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala b/core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala
@@ -851,11 +851,11 @@ private[spark] class TaskSetManager(
     val defaultWait = conf.get("spark.locality.wait", "3000ms")
     level match {
       case TaskLocality.PROCESS_LOCAL =>
-        Utils.timeStringToMs(conf.get("spark.locality.wait.process", defaultWait))
+        Utils.timeStringAsMs(conf.get("spark.locality.wait.process", defaultWait))
       case TaskLocality.NODE_LOCAL =>
-        Utils.timeStringToMs(conf.get("spark.locality.wait.node", defaultWait))
+        Utils.timeStringAsMs(conf.get("spark.locality.wait.node", defaultWait))
       case TaskLocality.RACK_LOCAL =>
-        Utils.timeStringToMs(conf.get("spark.locality.wait.rack", defaultWait))
+        Utils.timeStringAsMs(conf.get("spark.locality.wait.rack", defaultWait))
       case _ => 0L
     }
   }

diff --git a/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala b/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala
@@ -58,7 +58,7 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val actorSyste
   // Submit tasks after maxRegisteredWaitingTime milliseconds
   // if minRegisteredRatio has not yet been reached
   val maxRegisteredWaitingTimeMs =
-    Utils.timeStringToMs(conf.get("spark.scheduler.maxRegisteredResourcesWaitingTime", "30000ms"))
+    Utils.timeStringAsMs(conf.get("spark.scheduler.maxRegisteredResourcesWaitingTime", "30000ms"))
   val createTime = System.currentTimeMillis()
 
   private val executorDataMap = new HashMap[String, ExecutorData]
@@ -80,7 +80,7 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val actorSyste
       context.system.eventStream.subscribe(self, classOf[RemotingLifecycleEvent])
 
       // Periodically revive offers to allow delay scheduling to work
-      val reviveIntervalMs = Utils.timeStringToMs(
+      val reviveIntervalMs = Utils.timeStringAsMs(
         conf.get("spark.scheduler.revive.interval", "1000ms"))
       import context.dispatcher
       context.system.scheduler.schedule(0.millis, reviveIntervalMs.millis, self, ReviveOffers)

diff --git a/core/src/main/scala/org/apache/spark/util/AkkaUtils.scala b/core/src/main/scala/org/apache/spark/util/AkkaUtils.scala
@@ -66,7 +66,7 @@ private[spark] object AkkaUtils extends Logging {
 
     val akkaThreads   = conf.getInt("spark.akka.threads", 4)
     val akkaBatchSize = conf.getInt("spark.akka.batchSize", 15)
-    val akkaTimeoutS = Utils.timeStringToS(conf.get("spark.akka.timeout",
+    val akkaTimeoutS = Utils.timeStringAsS(conf.get("spark.akka.timeout",
       conf.get("spark.network.timeout", "120s")))
     val akkaFrameSize = maxFrameSizeBytes(conf)
     val akkaLogLifecycleEvents = conf.getBoolean("spark.akka.logLifecycleEvents", false)
@@ -81,7 +81,7 @@ private[spark] object AkkaUtils extends Logging {
 
     val akkaHeartBeatPauses = conf.getInt("spark.akka.heartbeat.pauses", 6000)
     val akkaHeartBeatIntervalS =
-      Utils.timeStringToS(conf.get("spark.akka.heartbeat.interval", "1000s"))
+      Utils.timeStringAsS(conf.get("spark.akka.heartbeat.interval", "1000s"))
 
     val secretKey = securityManager.getSecretKey()
     val isAuthOn = securityManager.isAuthenticationEnabled()

diff --git a/core/src/main/scala/org/apache/spark/util/MetadataCleaner.scala b/core/src/main/scala/org/apache/spark/util/MetadataCleaner.scala
@@ -76,7 +76,7 @@ private[spark] object MetadataCleanerType extends Enumeration {
 // initialization of StreamingContext. It's okay for users trying to configure stuff themselves.
 private[spark] object MetadataCleaner {
   def getDelaySeconds(conf: SparkConf): Int = {
-    Utils.timeStringToS(conf.get("spark.cleaner.ttl", "-1s")).toInt
+    Utils.timeStringAsS(conf.get("spark.cleaner.ttl", "-1s")).toInt
   }
 
   def getDelaySeconds(

diff --git a/core/src/main/scala/org/apache/spark/util/Utils.scala b/core/src/main/scala/org/apache/spark/util/Utils.scala
@@ -611,7 +611,7 @@ private[spark] object Utils extends Logging {
         }
         Utils.setupSecureURLConnection(uc, securityMgr)
 
-        val timeoutMs = Utils.timeStringToMs(conf.get("spark.files.fetchTimeout", "60s")).toInt
+        val timeoutMs = Utils.timeStringAsMs(conf.get("spark.files.fetchTimeout", "60s")).toInt
         uc.setConnectTimeout(timeoutMs)
         uc.setReadTimeout(timeoutMs)
         uc.connect()
@@ -1010,36 +1010,58 @@ private[spark] object Utils extends Logging {
     )
   }
 
+  /** Check whether a time-suffix was provided for the time string. */
+  private def hasTimeSuffix(str: String) : Boolean = {
+    val lower = str.toLowerCase.trim()
+    lower.endsWith("ms") || lower.endsWith("us") || lower.endsWith("s")
+  }
+
+  val timeError = "Time must be specified as seconds (s), " +
+      "milliseconds (ms), or microseconds (us) e.g. 50s, 100ms, or 250us."
+
   /**
-   * Convert a time parameter such as (50s, 100ms, or 250us) to microseconds for internal use
+   * Convert a passed time string (e.g. 50s, 100ms, or 250us) to a microsecond count for
+   * internal use. If no suffix is provided a direct conversion is attempted.
    */
-  def timeStringToUs(str: String): Long = {
-    val lower = str.toLowerCase.trim()
-    if (lower.endsWith("ms")) {
-      lower.substring(0, lower.length-2).toLong * 1000
-    } else if (lower.endsWith("us")) {
-      lower.substring(0, lower.length-2).toLong
-    } else if (lower.endsWith("s")) {
-      lower.substring(0, lower.length-1).toLong * 1000 * 1000
-    } else { // Invalid suffix, force correct formatting
-      throw new IllegalArgumentException("Time must be specified as seconds (s), " +
-          "milliseconds (ms), or microseconds (us) e.g. 50s, 100ms, or 250us.")
+  private def timeStringToUs(str: String) : Long = {
+    try {
+      val lower = str.toLowerCase.trim()
+      if (lower.endsWith("ms")) {
+        lower.substring(0, lower.length - 2).toLong * 1000
+      } else if (lower.endsWith("us")) {
+        lower.substring(0, lower.length - 2).toLong
+      } else if (lower.endsWith("s")) {
+        lower.substring(0, lower.length - 1).toLong * 1000 * 1000
+      } else {
+        // Invalid suffix, force correct formatting
+        lower.toLong
+      }
+    } catch {
+      case e: NumberFormatException => throw new NumberFormatException(timeError)
     }
   }
 
   /**
-   * Convert a time parameter such as (50s, 100ms, or 250us) to milliseconds for internal use.
-   * Note: may round in some cases
+   * Convert a time parameter such as (50s, 100ms, or 250us) to microseconds for internal use. If
+   * no suffix is provided, the passed number is assumed to be in us.
+   */
+  def timeStringAsUs(str: String): Long = {
+      timeStringToUs(str)
+  }
+
+  /**
+   * Convert a time parameter such as (50s, 100ms, or 250us) to microseconds for internal use. If
+   * no suffix is provided, the passed number is assumed to be in ms.
    */
-  def timeStringToMs(str : String) : Long = {
-    timeStringToUs(str)/1000
+  def timeStringAsMs(str : String) : Long = {
+      timeStringToUs(str)/1000
   }
 
   /**
-   * Convert a time parameter such as (50s, 100ms, or 250us) to seconds for internal use.
-   * Note: may round in some cases
+   * Convert a time parameter such as (50s, 100ms, or 250us) to microseconds for internal use. If
+   * no suffix is provided, the passed number is assumed to be in seconds.
    */
-  def timeStringToS(str : String) : Long = {
+  def timeStringAsS(str : String) : Long = {
     timeStringToUs(str)/1000/1000
   }
 

diff --git a/core/src/test/scala/org/apache/spark/scheduler/TaskSetManagerSuite.scala b/core/src/test/scala/org/apache/spark/scheduler/TaskSetManagerSuite.scala
@@ -152,7 +152,7 @@ class TaskSetManagerSuite extends FunSuite with LocalSparkContext with Logging {
 
   private val conf = new SparkConf
 
-  val LOCALITY_WAIT = Utils.timeStringToMs(conf.get("spark.locality.wait", "3000ms"))
+  val LOCALITY_WAIT = Utils.timeStringAsMs(conf.get("spark.locality.wait", "3000ms"))
   val MAX_TASK_FAILURES = 4
 
   override def beforeEach() {

diff --git a/streaming/src/main/scala/org/apache/spark/streaming/receiver/BlockGenerator.scala b/streaming/src/main/scala/org/apache/spark/streaming/receiver/BlockGenerator.scala
@@ -80,7 +80,7 @@ private[streaming] class BlockGenerator(
 
   private val clock = new SystemClock()
   private val blockIntervalMs =
-    Utils.timeStringToMs(conf.get("spark.streaming.blockIntervalMs", "200ms"))
+    Utils.timeStringAsMs(conf.get("spark.streaming.blockIntervalMs", "200ms"))
   private val blockIntervalTimer =
     new RecurringTimer(clock, blockIntervalMs, updateCurrentBuffer, "BlockGenerator")
   private val blockQueueSize = conf.getInt("spark.streaming.blockQueueSize", 10)

diff --git a/streaming/src/main/scala/org/apache/spark/streaming/scheduler/JobGenerator.scala b/streaming/src/main/scala/org/apache/spark/streaming/scheduler/JobGenerator.scala
@@ -104,7 +104,7 @@ class JobGenerator(jobScheduler: JobScheduler) extends Logging {
     if (processReceivedData) {
       logInfo("Stopping JobGenerator gracefully")
       val timeWhenStopStarted = System.currentTimeMillis()
-      val stopTimeout = Utils.timeStringToMs(conf.get(
+      val stopTimeout = Utils.timeStringAsMs(conf.get(
         "spark.streaming.gracefulStopTimeout", s"${10 * ssc.graph.batchDuration.milliseconds}ms"))
       val pollTime = 100
 

diff --git a/streaming/src/test/scala/org/apache/spark/streaming/StreamingContextSuite.scala b/streaming/src/test/scala/org/apache/spark/streaming/StreamingContextSuite.scala
@@ -75,7 +75,7 @@ class StreamingContextSuite extends FunSuite with BeforeAndAfter with Timeouts w
     val myConf = SparkContext.updatedConf(new SparkConf(false), master, appName)
     myConf.set("spark.cleaner.ttl", "10s")
     ssc = new StreamingContext(myConf, batchDuration)
-    assert(Utils.timeStringToS(ssc.conf.get("spark.cleaner.ttl", "-1s")) === 10)
+    assert(Utils.timeStringAsS(ssc.conf.get("spark.cleaner.ttl", "-1s")) === 10)
   }
 
   test("from existing SparkContext") {
@@ -87,7 +87,7 @@ class StreamingContextSuite extends FunSuite with BeforeAndAfter with Timeouts w
     val myConf = SparkContext.updatedConf(new SparkConf(false), master, appName)
     myConf.set("spark.cleaner.ttl", "10s")
     ssc = new StreamingContext(myConf, batchDuration)
-    assert(Utils.timeStringToS(ssc.conf.get("spark.cleaner.ttl", "-1s")) === 10)
+    assert(Utils.timeStringAsS(ssc.conf.get("spark.cleaner.ttl", "-1s")) === 10)
   }
 
   test("from checkpoint") {
@@ -98,12 +98,12 @@ class StreamingContextSuite extends FunSuite with BeforeAndAfter with Timeouts w
     ssc1.start()
     val cp = new Checkpoint(ssc1, Time(1000))
     assert(
-      Utils.timeStringToS(cp.sparkConfPairs.toMap.getOrElse("spark.cleaner.ttl", "-1s")) === 10)
+      Utils.timeStringAsS(cp.sparkConfPairs.toMap.getOrElse("spark.cleaner.ttl", "-1s")) === 10)
     ssc1.stop()
     val newCp = Utils.deserialize[Checkpoint](Utils.serialize(cp))
-    assert(Utils.timeStringToS(newCp.createSparkConf().get("spark.cleaner.ttl", "-1s")) === 10)
+    assert(Utils.timeStringAsS(newCp.createSparkConf().get("spark.cleaner.ttl", "-1s")) === 10)
     ssc = new StreamingContext(null, newCp, null)
-    assert(Utils.timeStringToS(ssc.conf.get("spark.cleaner.ttl", "-1s")) === 10)
+    assert(Utils.timeStringAsS(ssc.conf.get("spark.cleaner.ttl", "-1s")) === 10)
   }
 
   test("start and stop state check") {