Akka HTTPと長時間実行されるリクエスト

Question

私たちは、骨なしで実装されたAPIを持っています。Scala Akka HTTP - 重い計算（CPUとメモリを大量に消費する）に直面する2つのルート。クラスタ化はありません。すべてが1台の堅いマシンで実行されています。計算は重く、1つの独立した要求に対して60秒以上かかることがあります。そして、それほどスピードはあまり気にしません。ブロックIOはなく、多くのCPU処理があります。

パフォーマンステストを開始したとき、興味深いパターンが示されました：リクエストA1、A2、...、A10が来ると言う。彼らはリソースをかなり大量に使用しています.AkkaはoverranされたリクエストA5〜A10に対してHTTP 503を返します。問題は、結果を取り上げる人がいなくても計算がまだ実行されていることです。

そしてそこから、要求A11-A20が依頼依頼A5-A10に依然として対応しているサーバに到着します。明らかに、これらの新しい要求はまた、サーバがより混雑していることを考えると、それを上回る可能性があります。そのため、Akkaがタイムアウトを引き起こしてサーバーをより混雑させ、遅くしてから、新しいバッチのリクエストが届くようになるまでに、いくつかのサーバーが実行されます。システムを少し実行した後、特定のタイムアウトで失敗したポイントを開始します。また、負荷を停止すると、依然としてログに記録されているリクエストが表示されます。

私は計算を別のExecutionContextとシステムディスパッチャーで実行してみましたが、完全な非同期（未来のコンポジションを使用）にしようとしましたが、結果は変わりません。残念な仕事はサーバーを非常に忙しくして、最終的にはほとんどすべての要求が失敗します。

同様のケースはhttps://github.com/zcox/spray-blocking-testに記載されていますが、焦点はそこにシフトされています。/pingは、長時間実行される要求を処理するエンドポイントで多かれ少なかれ安定した責任があります。

質問：ハングしたリクエストを中断するのに適したアプリケーションを設計するにはどうすればよいですか？重い負荷のもとでは失敗した要求の一部を許容できますが、数秒後にシステム全体を停止させることは受け入れられません。

Answer 1

Akka HTTPは、タイムアウトした要求の処理を自動的に終了しません。通常、これを行うために必要となる余分な簿記は返済されないので、デフォルトではオンになっていません。私はそれが見落としのようなものだと思う、TBH、そして私はAkka HTTPに似た問題を自分で持っていた。

リクエストタイムアウトで処理を手動で中止する必要があると思います。そうしないと、見たとおりにサーバーが過負荷になっても回復しません。

これを実装できる標準的なメカニズムはありません（「How to cancel Future in Scala?」を参照）。スレッドがI/OなしのCPU作業を行っている場合、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が完了したかどうかの確認がされます

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