"リトライ"の実装: Truncated Binary Exponential Backoff #Scala

exponential backoff はシステムが許容可能な速度を見つけるために各処理の実行速度を遅延させるためのアルゴリズムです。簡単に言えば最適なリトライ ─ 例えば通信が切断された時に自動で再接続するまでの遅延時間を計算する方法のこと。exponential backoff は連続失敗回数を $n$ とした時に $0$ から $x^{n-1}$ までの間の乱数を使用する。

よく使われているアルゴリズムは $2$ のべき乗で計算する binary exponential backoff というもの。

$t = a \times 2^{n-1}$

$t$, $a$ を秒とした場合、1回目のリトライは $0～a$[sec]、2回目のリトライは $0～2a$[sec]、3回目のリトライは $0～4a$[sec]、... の間のランダムに選んだ時間をおく。

ただこの式だけでは $t$ がすぐに膨大な値になってしまうことが容易に想像がつくだろう。値がオーバーフローしたり、ネットワークが復旧したにも関わらずリトライが数ヶ月後に設定されていては自動復旧を実装する意味がないので $n$ に上限を与えて遅延時間がある一定値以上とならない条件を追加します。　

$n' = \min(n, N)$
$t = a \times 2^{n'-1}$
$T = a \times 2^{N-1}$

これを truncated binary exponential backoff と呼び、まぁここらへんが実用的な実装方針です。上限 $N$ と最大遅延時間 $T$ が決まれば $a$ を求めるのは簡単。

$a = \frac{T}{2^{N-1}}$

例えば「10回連続で失敗したら一時的・瞬間的な問題ではないためそれ以上は遅延を増加させない」「復旧後は10秒以内に再接続を実行する」と要件設定するのであれば $N=10$, $T=10$ から $a = 0.0195$[sec] で実装すれば良いことが分かります。

サンプルコード

Scala です。処理をブロックしても全体に影響が出ない Single Thread アプリであれば単純にループや再帰と sleep() を使用して再試行を実装すればよい。ここでシフト演算子が行っている 1 << (n-1) は $2^{n-1}$ と等価だが Int 幅の制約のため n=32 を超える場合は Long にするか math.pow(2, n-1) に変更する必要がある。

def retry[T](n:Int = 10, t:Long = 10000)(doSomething: =>T):T = {
  val a = t.toDouble / (1 << (n-1))
  @scala.annotation.tailrec
  def _retry(error:Int):T = try {
    doSomething
  } catch {
    case ex:Throwable => 
      val e = math.min(error+1, n)
      val interval = (a * (1 << (e-1)) * math.random).toInt
      println(f"$e%d: $interval%,d / ${(a * (1 << (e-1))).toInt}%,d")
      Thread.sleep(interval)
      _retry(e)
  }
  _retry(0)
}

// 13回目に成功する処理をリトライするまで実行する
var x:Int = 0
retry(){
  x += 1
  if(x==13) "hello" else throw new Exception()
}

サーバサイドなど多くの処理が計算リソースを共有している環境では非同期で実装した方が良いですね、ということで Timer と Future を使ったサンプル。timer スレッドを一つの処理で占有すると他の処理を巻き添えにするので実行スレッドを起動するのみに留めております。

import java.util.{Timer,TimerTask}
import scala.concurrent.{ExecutionContext,Future,Promise}
import scala.util.Try

val timer = new Timer(true)

def retry[T](n:Int = 10, t:Long = 10000)(doSomething: =>T)(implicit _ctx:ExecutionContext):Future[T] = {
  val a = t.toDouble / (1 << (n-1))
  val promise = Promise[T]()
  def _retry(error:Int):Unit = Future{
    try {
      promise.success(doSomething)
    }catch {
      case ex:Throwable => 
        val e = math.min(error+1, n)
        val interval = (a * (1 << (e-1)) * math.random).toInt
        println(f"$e%d: $interval%,d / ${(a * (1 << (e-1))).toInt}%,d")
        timer.schedule(new TimerTask(){
          def run() = _retry(e)
        }, interval)
    }
  }
  _retry(0)
  promise.future
}

// 13回目に成功する処理をリトライするまで実行する
import scala.concurrent.ExecutionContext.Implicits.global
var x:Int = 0
retry(){
  x += 1
  if(x==13) "hello" else throw new Exception()
}.onSuccess{ case msg => println(msg) }

実行すると想定通りにこんなものが表示されます。

1: 13 / 19
2: 19 / 39
3: 25 / 78
4: 35 / 156
5: 50 / 312
6: 216 / 625
7: 564 / 1,250
8: 635 / 2,500
9: 4,056 / 5,000
10: 7,238 / 10,000
10: 9,815 / 10,000
10: 6,159 / 10,000
hello