Akka HTTP Connection Pool Hangs After Couple of Ho

问题:

I have an HTTP Connection Pool that hangs after a couple of hours of running:

private def createHttpPool(host: String): SourceQueue[(HttpRequest, Promise[HttpResponse])] = {
    val pool = Http().cachedHostConnectionPoolHttps[Promise[HttpResponse]](host)
    Source.queue[(HttpRequest, Promise[HttpResponse])](config.poolBuffer, OverflowStrategy.dropNew)
      .via(pool).toMat(Sink.foreach {
        case ((Success(res), p)) => p.success(res)
        case ((Failure(e), p)) => p.failure(e)
      })(Keep.left).run
  }

I enqueue items with:

private def enqueue(uri: Uri): Future[HttpResponse] = {
    val promise = Promise[HttpResponse]
    val request = HttpRequest(uri = uri) -> promise

    queue.offer(request).flatMap {
      case Enqueued => promise.future
      case _ => Future.failed(ConnectionPoolDroppedRequest)
    }
}

And resolve the response like this:

private def request(uri: Uri): Future[HttpResponse] = {
    def retry = {
      Thread.sleep(config.dispatcherRetryInterval)
      logger.info(s"retrying")
      request(uri)
    }

    logger.info("req-start")
    for {
      response <- enqueue(uri)

      _ = logger.info("req-end")

      finalResponse <- response.status match {
        case TooManyRequests => retry
        case OK => Future.successful(response)
        case _ => response.entity.toStrict(10.seconds).map(s => throw Error(s.toString, uri.toString))
      }
    } yield finalResponse
}

The result of this function is then always transformed if the Future is successful:

def get(uri: Uri): Future[Try[JValue]] = {
  for {
    response <- request(uri)
    json <- Unmarshal(response.entity).to[Try[JValue]]
  } yield json
}

Everything works fine for a while and then all I see in the logs are req-start and no req-end.

My akka configuration is like this:

akka {
  actor.deployment.default {
    dispatcher = "my-dispatcher"
  }
}

my-dispatcher {
  type = Dispatcher
  executor = "fork-join-executor"

  fork-join-executor {
    parallelism-min = 256
    parallelism-factor = 128.0
    parallelism-max = 1024
  }
}

akka.http {
  host-connection-pool {
    max-connections = 512
    max-retries = 5
    max-open-requests = 16384
    pipelining-limit = 1
  }
}

I'm not sure if this is a configuration problem or a code problem. I have my parallelism and connection numbers so high because without it I get very poor req/s rate (I want to request as fast possible - I have other rate limiting code to protect the server).

回答1:

You are not consuming the entity of the responses you get back from the server. Citing the docs below:

Consuming (or discarding) the Entity of a request is mandatory! If accidentally left neither consumed or discarded Akka HTTP will assume the incoming data should remain back-pressured, and will stall the incoming data via TCP back-pressure mechanisms. A client should consume the Entity regardless of the status of the HttpResponse.

The entity comes in the form of a Source[ByteString, _] which needs to be run to avoid resource starvation.

If you don't need to read the entity, the simplest way to consume the entity bytes is to discard them, by using

res.discardEntityBytes()

(you can attach a callback by adding - e.g. - .future().map(...)).

This page in the docs describes all the alternatives to this, including how to read the bytes if needed.

--- EDIT

After more code/info was provided, it is clear that the resource consumption is not the problem. There is another big red flag in this implementation, namely the Thread.sleep in the retry method. This is a blocking call that is very likely to starve the threading infrastructure of your underlying actor system.

A full blown explanation of why this is dangerous was provided in the docs.

Try changing that and using akka.pattern.after (docs). Example below:

def retry = akka.pattern.after(200 millis, using = system.scheduler)(request(uri))