ScalaのFuture


Future　scalaで非同期演算を扱いたい時に使う
未完了、失敗、成功の状態がある
---

#基本

```scala
Future(Thread.sleep(1000 * 10)) // Future(<not completed>) 未完了
Future(1) // Future(Success(1))　成功 
Future(1/0) // Future(Failure(java.lang.ArithmeticException: / by zero)) 失敗、例外を保持
Future(throw new OutOfMemoryError) 
//java.lang.OutOfMemoryError 致命的な例外　ExecutionContextのreportFailureが呼ばれその後
//Thread.uncaughtexceptionhandlerが呼ばれる.　Futureは完了しない
```

---

※ Fatal（致命的）な例外とは以下の例外のもの

- VirtualMachineError,
  - OutOfMemoryとか
- ThreadDeath, 
    - スレッドが死んだらどうしようもない
- InterruptedException,
    - 割り込み要求されたら中断して投げるやつ
- LinkageError, 
    - AbstractMethodErrorなど
- ControlThrowable,
    - `Breakable.break`とか
- NotImplementedError


---


```scala

trait Future[+T] extends Awaitable[T]

```

Future traitに定義されているメソッドはAwaitableにあるreadyとresultメソッド以外は全部ノンブロッキング


---

## Futureのメソッドたち


---

## 成功, 失敗したら副作用を実行

成功時

```scala
Future(1).foreach(println) // １
```

失敗時

```scala
Future(1/0).failed.foreach(println) // java.lang.ArithmeticException: / by zero
```


＊　onSuccess, onFailureはScala 2.12から非奨励

---

## 成功も失敗も両方扱いたい(戻り値はUnit)

```scala
Future(1 / 0).andThen {
  case Success(x) => println(x)
  case Failure(_: ArithmeticException) => println("/ by zero")
} 
```

---

## 計算結果を使って処理したい

```scala
Future(1).map(_ + 1) // Future(Success(2))
```

途中で失敗したら後続の処理はskipしてそのまま結果が帰る

```scala
Future(1).map(_ / 0).map(_ + 1) // Future(Failure(java.lang.ArithmeticException: / by zero))
```
---

## 計算結果をFutureを返す関数に渡して合成したい

```scala
def futureFunc(i: Int): Future[Int] = ???
Future(1).flatMap(futureFunc) 
```

Futureを返す関数が複数合成したい時はfor式を使える

```scala
for {
 x1 <- futureFunc1()
 x2 <- futureFunc2(x1)
 x3 <- futureFunc3(x2)
} yield x3 
```
- ただしfutureFunc2はfutureFunc1が完了したあとに実行
futureFunc3はfutureFunc2が完了した後に実行される事に注意(並列に実行される訳ではない)


---

## Futureを並列に実行して合成したい

合成したいものが二つの場合

```scala
(Future(1) zip Future(2)).map { case (x, y) => x + y } // Future(Success(3))
```
scala 2.12では

```scala
(Future(1) zipWith Future(2))((x, y) => x + y) // Future(Success(3))
```
と書けるようになった

------

二つより多い時
scalazの|@|を使うと便利

```scala
import scalaz._
(Future(1) |@| Future(2) |@| Future(3))(_ + _ + _) // Future(Success(6))
```

catsではこう書く

```scala
import cats._
import cats.implicits._
(Future(1), Future(2), Future(3)).mapN(_ + _ + _) // Future(Success(6))
```
---------

個数が決まってないものはFuture.sequenceを使う

```scala
//Seq[Future[A]]を Future[Seq[A]]に変換 
Future.sequence(Seq(Future(1), Future(2), Future(3)))// Future(Success(List(1, 2, 3))
```

---

## 特定の例外を回復したい時

recover

```scala
Future(1 / 0).recover { 
  case _: ArithmeticException => 0  
}.map(_ + 1) //Future(Success(1))
```

---

## 成功も失敗も両方扱いたい(戻り値あり)

```scala
Future(1).transform(_ + 1, ex => ex)
```

このtransform　成功は成功　失敗は失敗にしか変換できず使いづらかったが
2.12以降は　

```scala
transform[T](f: Try[U] => Try[T])
```
とTryを受け取りTryを返せるようになったので

```scala
Future(1).transform { 
 case Success(i) => Success(i + 1)
 case Failure(e) => Success(0)
}
```

失敗　成功両方を扱えみたいに書けるようになり実用的になった

---

## Futureのコンパニオンオブジェクトに生えている便利メソッド達

---

# Future.successful

```scala
//Futureを実行したくないが型は合わせたい時
futureFunc1().flatMap { i => 
  if (i > 0) 
    futureFunc2(i)
  else 
    Future.successful(0)
}
```

---

# Future.sequence

```scala
//Seq[Future[A]]を Future[Seq[A]]に変換 
Future.sequence(Seq(Future(1), Future(2), Future(3)))// Future(Success(List(1, 2, 3))
```

```scala
//失敗したものあった時には失敗が帰る 
Future.sequence(Seq(Future(1), Future(throw new Exception), Future(3)))// Future(Failure(java.lang.Exception))
```

---

# Future.firstCompletedOf

```scala
//最初に完了したものを返す
Future.firstCompletedOf(Seq(Future { Thread.sleep(1000); 1 }, Future(2))) //Future(Success(2))
```

```scala
//最初に完了したものが失敗だったら失敗を返す
Future.firstCompletedOf(Seq(Future { Thread.sleep(1000); 1 }, Future(throw new Exception()))) //Future(Failure(java.lang.Exception))
```

---

# Future.find

```scala
//条件に合うものを返す
Future.find(List(Future { Thread.sleep(1000); 2 }, Future(4)))(_ % 2 == 0) //Future(Success(Some(2)))
```

```scala
//失敗したものは無視する
Future.find(List(Future { Thread.sleep(1000); 2 }, Future(throw new Exception)))(_ % 2 == 0) //Future(Success(Some(2)))
```

scala 2.12で　これまでのfind(以前は失敗を無視しないで報告していていた)は非奨励になった

---

# Future.traverse

```scala
//Seq(1, 2).map(Future(_)).sequnceと同じ
Future.traverse(Seq(1, 2))(Future(_)) // Future(Success(List(1, 2, 3))
```

---
scala 2.12から
fold, reduceが非奨励になり代わりにfoldLeft, reduceLeftが増えた。

----

# Future.unit


```scala
Future.successfull(())
```

を

```scala
Future.unit
```

と書けるようになった

----

また永遠に計算を完了しない

```scala
Future.never
```

もできた

----


---

# ExecutionContext

---


- Futureの各メソッドでimplicit parameterとして要求されるもの
- Futureは何を計算するかのWhatを表すがExecutionContextはどのように並列実行するかのHowを表す
- 基本スレッドプールのようなものでデフォルトはjavaのFork/Joinフレームワーク
  - `import scala.concurrent.ExecutionContext.Implicits.global`
  - akka actorのデフォルトもForkJoinPool

---

ExecutionContextの定義はシンプル
executeに渡されるRunnableがFutureに渡されていた計算
def execute(runnable: Runnable)　= runnable.runとかすると同期Futureとか作れる
＊prepareは無視していいです

```scala
trait ExecutionContext {

  def execute(runnable: Runnable): Unit

  
  def reportFailure(@deprecatedName('t) cause: Throwable): Unit

  
  def prepare(): ExecutionContext = this

}
```

---

#代表的なスレッドプール

---

固定スレッドプールの概念図

<img src="https://qiita-image-store.s3.amazonaws.com/0/88416/ec1f9758-2bef-550f-6e3e-06f9a85637a9.jpeg" width=60%>

キューに溜まったtaskを待機スレッドに渡し実行する　わかり易いモデル
しかし問題点も。。

---

- 並列数（＝スレッド）を増やすとtaskキューへのロックが多発して性能が劣化する
- スレッドを増やしても並列数はcpuのcore数分
- スレッドを増やしすぎるとコンテキストスイッチが多発して性能劣化(=C10K問題)

つまり　従来の１リクエスト＝１スレッドでは同時接続数に上限がある

---

そこでfork-join-pool


<img src="https://qiita-image-store.s3.amazonaws.com/0/88416/29432798-c556-d040-7052-333b570b0fc7.jpeg" width=60%>

- スレッド数がコア数×N個ほどなのでコンテキストスイッチが少ない
- 各スレッドがキューを持っているのでロック獲得待ちが少ない
- 暇なスレッドが出来たら他のスレッドからjobを奪う（work-stealing）のするのでバランスよくjobが振り分けられる
- １リクエスト＝１スレッドではないのでスレッド数の上限が同時接続数の上限ではなくなる！

---

無敵に見えるが向き不向きがある。。

---

問題

```scala
import scala.concurrent.ExecutionContext.Implicits.global
(1 to 8).map(_ => Future {Thread.sleep(1000 * 30)})
Future(1)
```

論理コアが8のcpuで最後の`Future(1)`の計算は最低で何秒かかるでしょうか

---

答え　最低でも30秒

---

- ForkJoinPoolのデフォルトのスレッド数が論理コア数なのでスレッドを論理コア数分ブロックしてしまうと
全体が止まる
- ForkJoinが得意なものはスレッドをブロックしない処理、 重い計算やブロッキングIOは基本的にやるべきではない
(例えばakka-actorだと全actorが止まってメッセージを処理出来なくってしまう)
- ここでいうブロッキング処理とはブロッキングIO、高価な計算、 Thread.sleep, synchronizedなどによるスレッドの資源獲得待ちのこと

---

といっても全てのIOがノンブロッキングではないし全ての計算をノンブロッキングな粒度まで分解できる訳ではない
ではForkJoinPoolでブロッキングする処理を扱いたい時にはどうするか

---

方法1
ForkJoinPool限定だがblockingメソッドでスレッドをブロックする処理を囲めばForkJoinPoolが必要と判断した場合に
Threadを新規に作りブロックしないようにしてくれる

```scala
import scala.concurrent._
(1 to 8).map(_ => Future(blocking {Thread.sleep(1000 * 30)}))
```

が新規にスレッドを作るのはコストが高いし必ずしもスレッドを作ってくれる訳ではない

---
方法２
王道　
固定スレッドプールのExecutionContextを作ってそこでブロックする処理を書く

```scala
val ec = ExecutionContext.fromExecutorService(Executors.newFixedThreadPool(num))
(1 to 8).map(_ => Future {Thread.sleep(1000 * 30)}(ec))
```

これでForkJoinを使用しているFutureはブロックされないようになる
しかし結局　固定スレッドプールをコンテキストスイッチが多発して性能は劣化する
（そもそも論としてマシンあたりの同時接続数をそこまで要求しない普通のwebアプリの場合固定スレッドプールの方が扱いやすいのかもしれない）

---
まとめ

- FutureはmapやflatMapを使って合成できる
- Future objectには便利なメソッドが生えている
- デフォルトのfork-joinスレッドプールをブロックしてはいけない
- ブロッキングをしたい時は使用するスレッドプールを吟味する
- Futureは用法を守って正しく服用する

---