[SPARK-7624] Revert "[SPARK-4939] revive offers periodically in LocalBackend" in 1.3 branch

This reverts commit e196da8.

Author: Davies Liu <davies@databricks.com>

Closes #6337 from davies/revert_revive and squashes the following commits:

be73f96 [Davies Liu] Revert "[SPARK-4939] revive offers periodically in LocalBackend"

Author: Davies Liu

core/src/main/scala/org/apache/spark/scheduler/local/LocalBackend.scala

@@ -19,8 +19,6 @@ package org.apache.spark.scheduler.local
 
 import java.nio.ByteBuffer
 
-import scala.concurrent.duration._
-
 import akka.actor.{Actor, ActorRef, Props}
 
 import org.apache.spark.{Logging, SparkContext, SparkEnv, TaskState}
@@ -48,8 +46,6 @@ private[spark] class LocalActor(
     private val totalCores: Int)
   extends Actor with ActorLogReceive with Logging {
 
-  import context.dispatcher   // to use Akka's scheduler.scheduleOnce()
-
   private var freeCores = totalCores
 
   private val localExecutorId = SparkContext.DRIVER_IDENTIFIER
@@ -78,16 +74,11 @@ private[spark] class LocalActor(
 
   def reviveOffers() {
     val offers = Seq(new WorkerOffer(localExecutorId, localExecutorHostname, freeCores))
-    val tasks = scheduler.resourceOffers(offers).flatten
-    for (task <- tasks) {
+    for (task <- scheduler.resourceOffers(offers).flatten) {
       freeCores -= scheduler.CPUS_PER_TASK
       executor.launchTask(executorBackend, taskId = task.taskId, attemptNumber = task.attemptNumber,
         task.name, task.serializedTask)
     }
-    if (tasks.isEmpty && scheduler.activeTaskSets.nonEmpty) {
-      // Try to reviveOffer after 1 second, because scheduler may wait for locality timeout
-      context.system.scheduler.scheduleOnce(1000 millis, self, ReviveOffers)
-    }
   }
 }