Idiomatically scheduling background work that dies with the main thread in Scala

Question

I have a scala program that runs for a while and then terminates. I'd like to provide a library to this program that, behind the scenes, schedules an asynchronous task to run every N seconds. I'd also like the program to terminate when the main entrypoint's work is finished without needing to explicitly tell the background work to shut down (since it's inside a library).

As best I can tell the idiomatic way to do polling or scheduled work in Scala is with Akka's ActorSystem.scheduler.schedule, but using an ActorSystem makes the program hang after main waiting for the actors. I then tried and failed to add another actor that joins on the main thread, seemingly because "Anything that blocks a thread is not advised within Akka"

I could introduce a custom dispatcher; I could kludge something together with a polling isAlive check, or adding a similar check inside each worker; or I could give up on Akka and just use raw Threads.

This seems like a not-too-unusual thing to want to do, so I'd like to use idiomatic Scala if there's a clear best way.

Answer 1

I don't think there is an idiomatic Scala way.

The JVM program terminates when all non-daemon thread are finished. So you can schedule your task to run on a daemon thread.

So just use Java functionality:

import java.util.concurrent._

object Main {

  def main(args: Array[String]): Unit = {

    // Make a ThreadFactory that creates daemon threads.
    val threadFactory = new ThreadFactory() {
      def newThread(r: Runnable) = {
        val t = Executors.defaultThreadFactory().newThread(r)
        t.setDaemon(true)
        t
      }
    }

    // Create a scheduled pool using this thread factory
    val pool = Executors.newSingleThreadScheduledExecutor(threadFactory)

    // Schedule some function to run every second after an initial delay of 0 seconds
    // This assumes Scala 2.12. In 2.11 you'd have to create a `new Runnable` manually
    // Note that scheduling will stop, if there is an exception thrown from the function
    pool.scheduleAtFixedRate(() => println("run"), 0, 1, TimeUnit.SECONDS)

    Thread.sleep(5000)
  }
}

You can also use guava to create a daemon thread factory with new ThreadFactoryBuilder().setDaemon(true).build().

Answer 2

If you use Akka scheduler you will be relying on highly tuned and optimized implementation that is well tested. Bringing up an actor system is a bit heavy weight though, I agree. Additionally you have to bring in a dependency on akka. If you are ok with that you can explicitly call system.shutdown from main when you are done, or wrap it in a function that will do it for you.

Alternatively, you could try something along these lines:

import scala.concurrent._
import ExecutionContext.Implicits.global

object Main extends App {

    def repeatEvery[T](timeoutMillis: Int)(f: => T): Future[T] = {
      val p = Promise[T]()
      val never = p.future
      f
      def timeout = Future {
        Thread.sleep(timeoutMillis)
        throw new TimeoutException
      }
      val failure = Future.firstCompletedOf(List(never, timeout))
      failure.recoverWith { case _ => repeatEvery(timeoutMillis)(f) }
    }

    repeatEvery(1000) {
      println("scheduled job called")
    }

    println("main started doing its work")
    Thread.sleep(10000)
    println("main finished")
}

Prints:

scheduled job called
main started doing its work
scheduled job called
scheduled job called
scheduled job called
scheduled job called
scheduled job called
scheduled job called
scheduled job called
scheduled job called
scheduled job called
main finished

I don't like that it uses Thread.sleep, but that is done to avoid using any other 3rd party schedulers and Scala Future does not provide timeout options. So you'll be wasting one thread on that scheduling task, but that's what Akka scheduler seems to do anyway. The difference is that perhaps you want a single scheduler for the whole JVM not to waste too many threads. The code I provided albeit simpler will waste a thread per job.