Akka HTTP и длительные запросы

Question

У нас есть API, реализованный в голой кости Scala Akka HTTP - пара маршрутов для тяжелых вычислений (интенсивность процессора и памяти). Нет кластеризации - все работает на одной мускулистой машине. Вычисление достаточно тяжелое - может потребоваться более 60 секунд для завершения одного изолированного запроса. И мы не очень-то заботимся о скорости. Там нет блокировки ввода-вывода, просто обработка процессора.

Когда я начал тестирование производительности, интересный образец показал: скажем, запросы A1, A2, ..., A10. Они довольно сильно используют ресурсы, и получается, что Akka вернет HTTP 503 для запросов A5-A10, которые перевернут. Проблема в том, что вычисление все еще работает, хотя никто не может найти результат.

И оттуда мы видим каскадное падение производительности: запросы A11-A20 поступают на сервер все еще работает над запросами A5-A10. Очевидно, что эти новые запросы также имеют шанс превышения - даже выше, если сервер более занят. Таким образом, некоторые из них будут работать к тому времени, когда Akka запускает тайм-аут, делая сервер еще более загруженным и медленным, а затем запускается новая партия запросов ... поэтому, немного запуская систему, вы видите, что почти все запросы после определенных точечный старт не работает с тайм-аутами. И после того, как вы остановите нагрузку, которую вы видите в журналах, некоторые запросы все еще обрабатываются.

Я попытался выполнить вычисление в отдельном ExecutionContext, а также диспетчер системы, пытаясь сделать его полностью асинхронным (через композицию Future), но результат все тот же. Затяжные задания делают сервер настолько занятым, что в итоге почти каждый запрос терпит неудачу.

Аналогичный случай описан в https://github.com/zcox/spray-blocking-test, но фокус смещен туда - /ping не имеет для нас значения, так как более или менее стабильная ответственность за конечную точку, которая обрабатывает длительные запросы.

Вопрос: как мне разработать мое приложение, чтобы лучше прерывать вешающие запросы? Я могу терпеть некоторый небольшой процент неудачных запросов при большой нагрузке, но прерывание всей системы до остановки через несколько секунд является неприемлемым.

Answer 1

Akka HTTP не автоматически прекращает обработку запросов, срок которых истекает. Обычно дополнительная бухгалтерия, которая необходима для этого, не окупается, поэтому она не включена по умолчанию. Я думаю, что это что-то вроде надзора, TBH, и у меня были аналогичные проблемы с самим Akka HTTP.

Я думаю, вам нужно вручную отменить обработку в тайм-аут запроса, иначе сервер не восстановится, когда он будет перегружен, как вы видели.

Существует не стандартный механизм, с помощью которого вы можете реализовать это (см. «How to cancel Future in Scala?»). Если поток работает с ЦП без ввода-вывода, то Thread.interrupt() не будет полезен. Вместо этого вы должны создать Deadline или Promise или аналогичный, который показывает, если запрос все еще открыт, и передать вокруг и периодически проверять тайм-аут во время вычисления:

// in the HTTP server class: 
val responseTimeout: Duration = 30.seconds 

val routes = 
    path("slowComputation") { 
    complete { 
     val responseTimeoutDeadline: Deadline = responseTimeout.fromNow 
     computeSlowResult(responseTimeoutDeadline) 
    } 
    } 

// in the processing code: 
def computeSlowResult(responseDeadline: Deadline): Future[HttpResponse] = Future { 
    val gatherInputs: List[_] = ??? 
    gatherInputs.fold(0) { (acc, next) => 

    // check if the response has timed out 
    if (responseDeadline.isOverdue()) 
     throw new TimeoutException() 

    acc + next // proceed with the calculation a little 
    } 
} 

(Проверка если Promise завершен будет много дешевле, чем проверка того, закончился ли срок действия Deadline. Я поставил код для последнего выше, поскольку его легче написать.)

Answer 2

spray-blocking-test использует библиотеки, которые, как я думаю, не существуют в Akka HTTP.Я бы с подобной проблемой, и я решил ее следующим образом:

application.conf

blocking-io-dispatcher { 
    type = Dispatcher 
    executor = "thread-pool-executor" 
    thread-pool-executor { 
    fixed-pool-size = 16 
    } 
    throughput = 1 
} 

---

Маршрут

complete { 
    Try(new URL(url)) match { 
    case scala.util.Success(u) => { 
     val src = Source.fromIterator(() => parseMovies(u).iterator) 

     src 
     .via(findMovieByTitleAndYear) 
     .via(persistMovies) 
     .completionTimeout(5.seconds) 
     .toMat(Sink.fold(Future(0))((acc, elem) => Applicative[Future].map2(acc, elem)(_ + _)))(Keep.right) 
     // run the whole graph on a separate dispatcher 
     .withAttributes(ActorAttributes.dispatcher("blocking-io-dispatcher")) 
     .run.flatten 
     .onComplete { 
      _ match { 
       case scala.util.Success(n) => logger.info(s"Created $n movies") 
       case Failure(t) => logger.error(t, "Failed to process movies") 
      } 
     } 

     Accepted 
    } 
    case Failure(t) => logger.error(t, "Bad URL"); BadRequest -> "Bad URL" 
    } 
} 

Ответ немедленно возвращается в то время как обработка сохраняет происходит в фоновом режиме.

Дополнительное чтение:

http://doc.akka.io/docs/akka/current/scala/dispatchers.html http://blog.akka.io/streams/2016/07/06/threading-and-concurrency-in-akka-streams-explained