Clever Geek Handbook

Akka http client blocked after max-open requests

I am taking this simple example from akka http doc: http://doc.akka.io/docs/akka-http/current/scala/http/client-side/request-level.html

And I modify it a bit to ask for one hundred requests. Application blocks after 32 requests (default setting max-open-requests

). Why?

import akka.actor.{Actor, ActorLogging, ActorSystem, Props}
import akka.http.scaladsl.Http
import akka.http.scaladsl.model._
import akka.stream.{ActorMaterializer, ActorMaterializerSettings}
import akka.util.ByteString

import scala.io.StdIn

object AkkaClientExample extends App {
  val system: ActorSystem = ActorSystem("BatchAkka")
  try {
    val unformattedAddresses = (1 to 100).map(i => s"Rue de la Gracieuse $i, Préverenges, Switzerland")

    val googleGeocoder = system.actorOf(GoogleGeocoder.props, "GoogleGeocoder")

    unformattedAddresses.foreach(e => googleGeocoder ! GoogleGeocoder.GeoCode(e))

    println(">>> Press ENTER to exit <<<")
    StdIn.readLine()
  } finally {
    system.terminate()
  }
}

object GoogleGeocoder {
  def props: Props = Props[GoogleGeocoder]

  final case class GeoCode(unformattedAddress: String)
}

class GoogleGeocoder extends Actor with ActorLogging {
  import GoogleGeocoder._
  import akka.pattern.pipe
  import context.dispatcher

  final implicit val materializer: ActorMaterializer = ActorMaterializer(ActorMaterializerSettings(context.system))

  val http = Http(context.system)  

  def receive = {
    case GeoCode(unformattedAddress) =>
      log.info(s"GeoCode $unformattedAddress")
      http
        .singleRequest(HttpRequest(uri = url(unformattedAddress)))
        .map(r => (unformattedAddress, r))
        .pipeTo(self)

    case (unformattedAddress: String, resp @ HttpResponse(StatusCodes.OK, headers, entity, _)) =>
      log.info(s"Success response comming for $unformattedAddress")
      entity.dataBytes.runFold(ByteString(""))(_ ++ _).foreach { body =>
        val response = body.utf8String.replaceAll("\\s+", " ").take(50)
        log.info(s"Success response for $unformattedAddress: $response")
      }

    case (unformattedAddress: String, resp @ HttpResponse(code, _, _, _)) =>
      log.info(s"Request failed, response code: $code for $unformattedAddress")
      resp.discardEntityBytes()
  }

  def url(unformattedAddress: String): String =
    //s"https://maps.googleapis.com/maps/api/geocode/json?address=${URLEncoder.encode(unformattedAddress, "UTF-8")}&key=${URLEncoder.encode(googleApiKey, "UTF-8")}"
    s"https://www.epfl.ch/"
}

output:

[INFO] [07/28/2017 20:08:26.977] [BatchAkka-akka.actor.default-dispatcher-4] [akka://BatchAkka/user/GoogleGeocoder] GeoCode Rue de la Gracieuse 1, Préverenges, Switzerland
[INFO] [07/28/2017 20:08:27.080] [BatchAkka-akka.actor.default-dispatcher-4] [akka://BatchAkka/user/GoogleGeocoder] GeoCode Rue de la Gracieuse 2, Préverenges, Switzerland
...
[INFO] [07/28/2017 20:08:27.098] [BatchAkka-akka.actor.default-dispatcher-13] [akka://BatchAkka/user/GoogleGeocoder] GeoCode Rue de la Gracieuse 99, Préverenges, Switzerland
[INFO] [07/28/2017 20:08:27.098] [BatchAkka-akka.actor.default-dispatcher-13] [akka://BatchAkka/user/GoogleGeocoder] GeoCode Rue de la Gracieuse 100, Préverenges, Switzerland

[INFO] [07/28/2017 20:08:27.615] [BatchAkka-akka.actor.default-dispatcher-11] [akka://BatchAkka/user/GoogleGeocoder] Success response comming for Rue de la Gracieuse 1, Préverenges, Switzerland
[INFO] [07/28/2017 20:08:27.620] [BatchAkka-akka.actor.default-dispatcher-11] [akka://BatchAkka/user/GoogleGeocoder] Success response comming for Rue de la Gracieuse 4, Préverenges, Switzerland
[INFO] [07/28/2017 20:08:27.668] [BatchAkka-akka.actor.default-dispatcher-17] [akka://BatchAkka/user/GoogleGeocoder] Success response for Rue de la Gracieuse 4, Préverenges, Switzerland: <!doctype html><html lang="fr" class="no-js"><head
[INFO] [07/28/2017 20:08:27.668] [BatchAkka-akka.actor.default-dispatcher-21] [akka://BatchAkka/user/GoogleGeocoder] Success response for Rue de la Gracieuse 1, Préverenges, Switzerland: <!doctype html><html lang="fr" class="no-js"><head
...
[INFO] [07/28/2017 20:08:27.787] [BatchAkka-akka.actor.default-dispatcher-5] [akka://BatchAkka/user/GoogleGeocoder] Success response comming for Rue de la Gracieuse 31, Préverenges, Switzerland
[INFO] [07/28/2017 20:08:27.795] [BatchAkka-akka.actor.default-dispatcher-15] [akka://BatchAkka/user/GoogleGeocoder] Success response comming for Rue de la Gracieuse 32, Préverenges, Switzerland
[INFO] [07/28/2017 20:08:27.802] [BatchAkka-akka.actor.default-dispatcher-16] [akka://BatchAkka/user/GoogleGeocoder] Success response for Rue de la Gracieuse 31, Préverenges, Switzerland: <!doctype html><html lang="fr" class="no-js"><head
[INFO] [07/28/2017 20:08:27.806] [BatchAkka-akka.actor.default-dispatcher-17] [akka://BatchAkka/user/GoogleGeocoder] Success response for Rue de la Gracieuse 32, Préverenges, Switzerland: <!doctype html><html lang="fr" class="no-js"><head

blocked after the first 32 requests.

Update based on @shutty answer:

I modified the program as follows and it works:

class GoogleGeocoder extends Actor with ActorLogging {
  import GoogleGeocoder._
  import akka.pattern.pipe
  import context.dispatcher

  final implicit val materializer: ActorMaterializer = ActorMaterializer(ActorMaterializerSettings(context.system))

  val http = Http(context.system)  

  val queue = new scala.collection.mutable.Queue[String]
  var currentRequests = 0
  val MaxCurrentRequest = 10

  def receive = {
    case GeoCode(unformattedAddress) =>
      if (currentRequests < MaxCurrentRequest)
        query(unformattedAddress)
      else
        queue += unformattedAddress

    case (unformattedAddress: String, resp @ HttpResponse(StatusCodes.OK, headers, entity, _)) =>
      log.info(s"Success response comming for $unformattedAddress")
      entity.dataBytes.runFold(ByteString(""))(_ ++ _).foreach { body =>
        currentRequests = currentRequests - 1
        queryNext()
        val response = body.utf8String.replaceAll("\\s+", " ").take(50)
        log.info(s"Success response for $unformattedAddress: $response")
      }

    case (unformattedAddress: String, resp @ HttpResponse(code, _, _, _)) =>
      log.info(s"Request failed, response code: $code for $unformattedAddress")
      resp.discardEntityBytes()
      currentRequests = currentRequests - 1
      queryNext()

    case f: Status.Failure =>
      log.info("failure" + textSample(f))

    case m =>
      log.info("unexpected message: " + textSample(m))
  }

  def query(unformattedAddress: String) {
    log.info(s"GeoCode $unformattedAddress")
    http
      .singleRequest(HttpRequest(uri = url(unformattedAddress)))
      .map(r => (unformattedAddress, r))
      .pipeTo(self)
  }

  def queryNext() {
    if (queue.nonEmpty) {
      query(queue.dequeue)
    }
  }

  def url(unformattedAddress: String): String =
    //s"https://maps.googleapis.com/maps/api/geocode/json?address=${URLEncoder.encode(unformattedAddress, "UTF-8")}&key=${URLEncoder.encode(googleApiKey, "UTF-8")}"
    s"https://www.epfl.ch/"
}

So basically adding a queue.

However, is there a better way to achieve this?

I imagine cases where this implementation might fail: for example, if it http.singleRequest

creates a bad future, currentRequests

it won't shrink. I could handle this on case f: Status.Failure

, but still this solution looks very error prone.

Maybe akka already provides some mechanism for handling the queue?

Is there a way to add back pressure to the client (so AkkaClientExample

: unformattedAddresses.foreach(e => googleGeocoder ! GoogleGeocoder.GeoCode(e))

block on reaching MaxCurrentRequest

)?

+3


source to share


1 answer


If you run your example with akka.logging = DEBUG

, you will see the following output:

InputBuffer (max-open-requests = 32) now filled with 31 request after enqueuing GET / Empty InputBuffer (max-open-requests = 32) now filled with 32 request after enqueuing GET / Empty InputBuffer (max-open-requests = 32) exhausted when trying to enqueue GET / Empty InputBuffer (max-open-requests = 32) exhausted when trying to enqueue GET / Empty InputBuffer (max-open-requests = 32) exhausted when trying to enqueue GET / Empty

There is a pretty detailed description of how akka-http handles pooling client requests, but in short, if you overload the pool more than thant max-open-requests it will start dropping requests:

http
 .singleRequest(HttpRequest(uri = url(unformattedAddress)))
 .map(r => (unformattedAddress, r)) // <- HERE
 .pipeTo(self)

When you make a map through the future in Scala, it will only execute your callback on success in the future, which is not consistent with your code. If you rewrite your code a little differently, for example:

http
  .singleRequest(HttpRequest(uri = url(unformattedAddress)))
  .onComplete {
    case Success(r) =>
      self ! (unformattedAddress, r)
    case Failure(ex) =>
      log.error(ex, "pool overflow")
  }


You will see a bunch of exceptions complaining about a bad future.

Updated:

As for my own opinion, actors and streams are not very suitable when you need backpressure. As an option, you can completely rewrite your code without actors:

def url(addr: String) = "http://httpbin.org/headers"
implicit val system: ActorSystem = ActorSystem("BatchAkka")
implicit val mat: ActorMaterializer = ActorMaterializer()
import system.dispatcher
val http = Http()
val addresses = (1 to 100).map(i => s"Rue de la Gracieuse $i, Préverenges, Switzerland")
Source(addresses)
  .mapAsync(4)(addr => http.singleRequest(HttpRequest(uri = url(addr))))
  .map(response => println(response.status))
  .runWith(Sink.seq)
  .map(_ => println("done"))

In this solution, you will only have 4 concurrent requests to the server with back pressure, calls and whistles.

+2


source






More articles:


All Articles