Potential bug in managing number of connections in akka-http server

[akka-ci]

Issue by sjulias
Monday May 23, 2016 at 21:24 GMT
Originally opened as akka/akka#20600

---

I'm evaluating akka-http
dependencies:

  scalaVersion := "2.11.7"
  "com.typesafe.akka" %% "akka-actor" % "2.4.6",
  "com.typesafe.akka" % "akka-http-experimental_2.11" % "2.4.6"

The code look like as following:

   object StatsRepo{
     val totConn = new AtomicInteger(0)
     val concReq = new AtomicInteger(0)
     val currOpenConn = new AtomicInteger(0) // how to count this?
   }
  object CommonRepo{
    implicit val system = ActorSystem("akka-http")
    val materializer = ActorMaterializer()
    val executionContext = system.dispatcher
    val workerExecutionContext = system.dispatchers.lookup("worker-dispatcher")
  }

  object Boot2 extends App{
      val requestHandler: HttpRequest => Future[HttpResponse] = {
        // do some work here...
           case HttpRequest(POST, url, headers, entity, _) => {
               StatsRepo.totReq.incrementAndGet()
               StatsRepo.concReq.incrementAndGet()
                // ...
                implicit val ec = CommonRepo.workerExecutionContext
               val bodyF: Future[String] = Unmarshal(entity).to[String]
               val resp = processRequestAsync(bodyF)
               convertToHttpResponseAsync(resp)
         }
     }

   def processRtbRequestAsync( bodyF: Future[String]): Future[MyRequestContext] = {
        implicit val ec = CommonRepo.workerExecutionContext
        extractBody(bodyF).
       withFilter(ctx=>ctx.body != null ).
       flatMap(ctx=> callToFunc1(ctx)). // func1:MyRequestContext->Future[MyRequestContext]
       flatMap(ctx => callToFunc2(ctx)) // func2 similar to func1 app code   
      // both funcs does not execute any blockind code, for test purpose I event put some immediate         dummy response there, like Future{ctx}
   }

def convertToHttpResponseAsync(resp: Future[MyRequestContext]): Future[HttpResponse] = {
    implicit val ec = BidderDIRepo.repo.clientTP
   resp.map { ctx =>
        StatsRepo.concReq.decrementAndGet()
       HttpResponse(StatusCodes.OK, entity = HttpEntity(`application/json`, ctx.resp))
    }
}

  implicit val system = ActorSystem("akka-http")
  implicit val materializer = ActorMaterializer()
  implicit val executionContext = system.dispatcher

  val serverSource = Http().bind("0.0.0.0", 8080)
  val bindingFuture: Future[Http.ServerBinding] =
  serverSource.to(Sink.foreach { connection =>
              StatsRepo.totConn.incrementAndGet()
              connection handleWithAsyncHandler requestHandler
   }).run()
     println(s"Server online at http://0.0.0.0:8080")
   }

In the the application.conf:

  http {
        server {
            max-connections = 10
            request-timeout = 50ms
            pipelining-limit = 1
       }
    }
  workers-dispatcher {
  type = Dispatcher
  executor = "thread-pool-executor"
  thread-pool-executor {
          core-pool-size-min = 5
          core-pool-size-factor = 0
          core-pool-size-max = 5
   }
  throughput = 1
  }

The issue I see in production, that in-spite of the max-connection limit the server has much more open-connections and processing big amount of concurrent requests
In addition such behavior could be reproduced with ab
If you run two windows with

  ab -k -n 100000 -c 8 -p ~/my.json  -T 'application/json' http://localhost:8080/test

In this case I'm getting ~15 concurent requests that also comes in spiky behavior
Also the latencies has spikes between start, func1,func2 processing... (but those futures are with already dummy data...?)