Sequencing `Future`s with timeout

Question

I utilized the TimeoutScheduler introduced at Scala Futures - built in timeout?.

However, now my program does not terminate as before without TimeoutScheduler.

I have two Futures: res1 and res2. Both with a timeout of 15 seconds. In the end I sequence both Futures in order to shutdown the HTTP executor properly in the onComplete callback. Without using withTimeout the program terminates right after http.shutdown. But with using withTimeout is doesn't. Why? There must be some further futures...

import java.net.URI
import scala.util.{ Try, Failure, Success }
import dispatch._
import org.json4s._
import org.json4s.native.JsonMethods._
import com.typesafe.scalalogging.slf4j.Logging

object Main extends App with Logging {
  import scala.concurrent.ExecutionContext.Implicits.global

  val http   = Http
  val github = host("api.github.com").secure 

  import timeout._
  import scala.concurrent.duration._
  import scala.language.postfixOps
  import scala.collection.JavaConverters._

  val req: dispatch.Req =  github / "users" / "defunkt"
  val res1 = http(req > dispatch.as.Response(_.getHeaders().get("Server").asScala)) withTimeout (15 seconds) recover { case x => 
    logger.debug("Error1: " + x.toString) 
    Nil
  }
  val res2 = http(req > dispatch.as.Response(_.getHeaders().get("Vary").asScala)) withTimeout (15 seconds) recover { case x => 
    logger.debug("Error2: " + x.toString) 
    Nil
  }

  Future.sequence(res1 :: res2 :: Nil) onComplete { case _ => 
    http.shutdown()               // without using `withTimeout` the program terminated after `http.shutdow`
    TimeoutScheduler.timer.stop() // thanks to @cmbaxter
  }
}

object timeout {
  import java.util.concurrent.TimeUnit
  import scala.concurrent.Promise
  import scala.concurrent.duration.Duration
  import org.jboss.netty.util.Timeout
  import org.jboss.netty.util.TimerTask
  import org.jboss.netty.util.HashedWheelTimer
  import org.jboss.netty.handler.timeout.TimeoutException

   // cf. https://stackoverflow.com/questions/16304471/scala-futures-built-in-timeout
  object TimeoutScheduler {
    val timer = new HashedWheelTimer(10, TimeUnit.MILLISECONDS)
    def scheduleTimeout(promise: Promise[_], after: Duration) = {
      timer.newTimeout(new TimerTask{
        def run(timeout: Timeout){              
          promise.failure(new TimeoutException("Operation timed out after " + after.toMillis + " millis"))        
        }
      }, after.toNanos, TimeUnit.NANOSECONDS)
    }
  }
  implicit class FutureWithTimeout[T](f: Future[T]) {
    import scala.concurrent.ExecutionContext

    def withTimeout(after: Duration)(implicit ec: ExecutionContext) = {
      val prom = Promise[T]()
      val timeout = TimeoutScheduler.scheduleTimeout(prom, after)
      val combinedFut = Future.firstCompletedOf(List(f, prom.future))
      f onComplete { case result => timeout.cancel() }
      combinedFut
    }
  } 
}

Any suggestions are welcome, Best, /nm

cmbaxter · Accepted Answer · 2014-02-24T18:38:20.557

4

If you used my code exactly as described, then my guess is that the Netty HashedWheelTimer that sits under the TimeoutScheduler object is not being terminated. You could try explicitly calling stop on it, after calling http.shutdown like so:

TimeoutScheduler.timer.stop

Now if you wanted the Netty HashedWheelTimer to use a daemon thread, then you can use one of the constructors on it (I'm seeing them in 3.6.6-Final) that accepts a ThreadFactory and then use a custom ThreadFactory that sets the daemon flag to true.

edited Feb 24 '14 at 18:38

answered Feb 24 '14 at 12:59

cmbaxter

35,283
4
86
95

Well, I only slightly modified your code. Do you think this is the matter? The entire code is in my initial post. Yours is at the very end. Or maybe `HashedWheelTimer` is not daemonic (suggested by Viktor Klang)? – nemron Feb 24 '14 at 15:08
Yes, the `HashedWheelTimer` thread is non-daemon, so you will have to explicitly stop it if you want to use that class as your efficient timer mechanism. There might be different options out there for a timer class though, so you could also look into that. Me using `HashedWheelTimer` was just for completeness of example. Don't feel that you have to use it if you don't like explicitly having to shut it down. – cmbaxter Feb 24 '14 at 16:31
Thanks for your reply. I am fine with explicitly shutting it down. Just was curious if I really have to. Thanks for the clarification. Last but not least, for didactical purposes, do you know an daemonic alternative I could play with, just to get a better understanding? Best, /nm – nemron Feb 24 '14 at 17:05
It looks like I may have spoken too soon. It appears that the Netty `HashedWheelTimer` supports a constructor that accepts a `ThreadFactory` that is used to spawn the thread that represents the timer. If you use that constructor, you should be able to pass in a custom `ThreadFactory` that sets the daemon flag on the `Thread` returned to true. – cmbaxter Feb 24 '14 at 18:36
Thank you for your help and investigation! You are right! I changed to `new HashedWheelTimer(new DefaultDaemonicThreadFactory, 10, TimeUnit.MILLISECONDS)` whereas `DefaultDaemonicThreadFactory` is a primitive copy of `Executors#DefaultThreadFactory` but with `t.setDaemon(true)`. It works! `TimeoutScheduler.timer.stop()`, in this case, is not necessary. – nemron Feb 25 '14 at 07:02

Sequencing `Future`s with timeout

1 Answers1

Linked