make retry() more flexible (JuliaLang#19331)

base/error.jl

@@ -82,45 +82,70 @@ macro assert(ex, msgs...)
  return :($(esc(ex)) ? $(nothing) : throw(Main.Base.AssertionError($msg)))
 end
 
-# NOTE: Please keep the constant values specified below in sync with the doc string
-const DEFAULT_RETRY_N = 1
-const DEFAULT_RETRY_ON = e->true
-const DEFAULT_RETRY_MAX_DELAY = 10.0
+immutable ExponentialBackOff
+ n::Int
+ first_delay::Float64
+ max_delay::Float64
+ factor::Float64
+ jitter::Float64
+
+ function ExponentialBackOff(n, first_delay, max_delay, factor, jitter)
+ all(x->x>=0, (n, first_delay, max_delay, factor, jitter)) || error("all inputs must be non-negative")
+ new(n, first_delay, max_delay, factor, jitter)
+ end
+end
 
 """
- retry(f, [retry_on]; n=1, max_delay=10.0) -> Function
+ ExponentialBackOff(; n=1, first_delay=0.05, max_delay=10.0, factor=5.0, jitter=0.1)
 
-Returns a lambda that retries function `f` up to `n` times in the
-event of an exception. If `retry_on` is a `Type` then retry only
-for exceptions of that type. If `retry_on` is a function
-`test_error(::Exception) -> Bool` then retry only if it is true.
+A `Float64` iterator of length `n` whose elements exponentially increase at a
+rate in the interval `factor` * (1 ± `jitter`). The first element is
+`first_delay` and all elements are clamped to `max_delay`.
+"""
+ExponentialBackOff(; n=1, first_delay=0.05, max_delay=10.0, factor=5.0, jitter=0.1) =
+ ExponentialBackOff(n, first_delay, max_delay, factor, jitter)
+start(ebo::ExponentialBackOff) = (ebo.n, min(ebo.first_delay, ebo.max_delay))
+function next(ebo::ExponentialBackOff, state)
+ next_n = state[1]-1
+ curr_delay = state[2]
+ next_delay = min(ebo.max_delay, state[2] * ebo.factor * (1.0 - ebo.jitter + (rand() * 2.0 * ebo.jitter)))
+ (curr_delay, (next_n, next_delay))
+end
+done(ebo::ExponentialBackOff, state) = state[1]<1
+length(ebo::ExponentialBackOff) = ebo.n
 
-The first retry happens after a gap of 50 milliseconds or `max_delay`,
-whichever is lower. Subsequently, the delays between retries are
-exponentially increased with a random factor up to `max_delay`.
+"""
+ retry(f::Function; delays=Base.ExponentialBackOff(), check=nothing) -> Function
+
+Returns an anonymous function that calls function `f`. If an exception arises,
+`f` is repeatedly called again, each time `check` returns `true`, after waiting the
+number of seconds specified in `delays`. `check` should input `delays`'s
+current state and the `Exception`.
 
-**Examples**
+# Examples
 ```julia
-retry(http_get, e -> e.status == "503")(url)
-retry(read, UVError)(io)
+retry(f, delays=fill(5.0, 3))
+retry(f, delays=rand(5:10, 2))
+retry(f, delays=Base.ExponentialBackOff(n=3, first_delay=5, max_delay=1000))
+retry(http_get, check=(s,e)->e.status == "503")(url)
+retry(read, check=(s,e)->isa(e, UVError))(io)
 ```
 """
-function retry(f::Function, retry_on::Function=DEFAULT_RETRY_ON; n=DEFAULT_RETRY_N, max_delay=DEFAULT_RETRY_MAX_DELAY)
+function retry(f::Function; delays=ExponentialBackOff(), check=nothing)
  (args...) -> begin
- delay = min(0.05, max_delay)
- for i = 1:n+1
+ state = start(delays)
+ while true
  try
  return f(args...)
  catch e
- if i > n || try retry_on(e) end !== true
- rethrow(e)
+ done(delays, state) && rethrow(e)
+ if check !== nothing
+ state, retry_or_not = check(state, e)
+ retry_or_not || rethrow(e)
  end
  end
- delay = min(max_delay, delay)
- sleep(delay * (0.8 + (rand() * 0.2)))
- delay = delay * 5
+ (delay, state) = next(delays, state)
+ sleep(delay)
  end
  end
 end
-
-retry(f::Function, t::Type; kw...) = retry(f, e->isa(e, t); kw...)

base/exports.jl

@@ -59,6 +59,7 @@ export
  EachLine,
  Enum,
  Enumerate,
+ ExponentialBackOff,
  Factorization,
  FileMonitor,
  FloatRange,

base/pmap.jl

@@ -30,7 +30,7 @@ pgenerate(f, c) = pgenerate(default_worker_pool(), f, c)
 pgenerate(f, c1, c...) = pgenerate(a->f(a...), zip(c1, c...))
 
 """
- pmap([::AbstractWorkerPool], f, c...; distributed=true, batch_size=1, on_error=nothing, retry_n=0, retry_max_delay=DEFAULT_RETRY_MAX_DELAY, retry_on=DEFAULT_RETRY_ON) -> collection
+ pmap([::AbstractWorkerPool], f, c...; distributed=true, batch_size=1, on_error=nothing, retry_delays=[]), retry_check=nothing) -> collection
 
 Transform collection `c` by applying `f` to each element using available
 workers and tasks.
@@ -59,27 +59,27 @@ you can specify an error handling function via argument `on_error` which takes i
 the exception. The function can stop the processing by rethrowing the error, or, to continue, return any value
 which is then returned inline with the results to the caller.
 
-Failed computation can also be retried via `retry_on`, `retry_n`, `retry_max_delay`, which are passed through
-to `retry` as arguments `retry_on`, `n` and `max_delay` respectively. If batching is specified, and an entire batch fails,
-all items in the batch are retried.
+Failed computation can also be retried via `retry_delays`, `retry_check`, which
+are passed through to `retry` as keyword arguments `delays` and `check`,
+respectively. If batching is specified, and an entire batch fails, all items in
+the batch are retried.
 
 The following are equivalent:
 
 * `pmap(f, c; distributed=false)` and `asyncmap(f,c)`
-* `pmap(f, c; retry_n=1)` and `asyncmap(retry(remote(f)),c)`
-* `pmap(f, c; retry_n=1, on_error=e->e)` and `asyncmap(x->try retry(remote(f))(x) catch e; e end, c)`
+* `pmap(f, c; retry_delays=Base.ExponentialBackOff())` and `asyncmap(retry(remote(f)),c)`
+* `pmap(f, c; retry_delays=Base.ExponentialBackOff(), on_error=e->e)` and `asyncmap(x->try retry(remote(f))(x) catch e; e end, c)`
 """
 function pmap(p::AbstractWorkerPool, f, c; distributed=true, batch_size=1, on_error=nothing,
- retry_n=0,
- retry_max_delay=DEFAULT_RETRY_MAX_DELAY,
- retry_on=DEFAULT_RETRY_ON,
+ retry_delays=[],
+ retry_check=nothing,
  # deprecated keyword args:
  err_retry=nothing, err_stop=nothing, pids=nothing)
  #15409
  if err_retry !== nothing
  depwarn("err_retry is deprecated, use pmap(retry(f), c...).", :pmap)
  if err_retry == true
- f = retry(f)
+ f = retry(f, delays=retry_delays, check=retry_check)
  end
  end
  if pids !== nothing
@@ -110,8 +110,8 @@ function pmap(p::AbstractWorkerPool, f, c; distributed=true, batch_size=1, on_er
  f = remote(p, f)
  end
 
- if retry_n > 0
- f = wrap_retry(f, retry_on, retry_n, retry_max_delay)
+ if length(retry_delays) > 0
+ f = wrap_retry(f, retry_delays, retry_check)
  end
  if on_error !== nothing
  f = wrap_on_error(f, on_error)
@@ -122,18 +122,18 @@ function pmap(p::AbstractWorkerPool, f, c; distributed=true, batch_size=1, on_er
  # During batch processing, We need to ensure that if on_error is set, it is called
  # for each element in error, and that we return as many elements as the original list.
  # retry, if set, has to be called element wise and we will do a best-effort
- # to ensure that we do not call mapped function on the same element more than retry_n.
+ # to ensure that we do not call mapped function on the same element more than length(retry_delays).
  # This guarantee is not possible in case of worker death / network errors, wherein
  # we will retry the entire batch on a new worker.
- if (on_error !== nothing) || (retry_n > 0)
+ if (on_error !== nothing) || (length(retry_delays) > 0)
  f = wrap_on_error(f, (x,e)->BatchProcessingError(x,e); capture_data=true)
  end
  f = wrap_batch(f, p, on_error)
  results = asyncmap(f, c; ntasks=()->nworkers(p), batch_size=batch_size)
 
  # handle error processing....
- if (on_error !== nothing) || (retry_n > 0)
- process_batch_errors!(p, f_orig, results, on_error, retry_on, retry_n, retry_max_delay)
+ if (on_error !== nothing) || (length(retry_delays) > 0)
+ process_batch_errors!(p, f_orig, results, on_error, retry_delays, retry_check)
  end
 
  return results
@@ -158,7 +158,7 @@ function wrap_on_error(f, on_error; capture_data=false)
  end
 end
 
-wrap_retry(f, retry_on, n, max_delay) = retry(f, retry_on; n=n, max_delay=max_delay)
+wrap_retry(f, retry_delays, retry_check) = retry(f, delays=retry_delays, check=retry_check)
 
 function wrap_batch(f, p, on_error)
  f = asyncmap_batch(f)
@@ -177,9 +177,9 @@ end
 
 asyncmap_batch(f) = batch -> asyncmap(x->f(x...), batch)
 
-function process_batch_errors!(p, f, results, on_error, retry_on, retry_n, retry_max_delay)
+function process_batch_errors!(p, f, results, on_error, retry_delays, retry_check)
  # Handle all the ones in error in another pmap, with batch size set to 1
- if (on_error !== nothing) || (retry_n > 0)
+ if (on_error !== nothing) || (length(retry_delays) > 0)
  reprocess = []
  for (idx, v) in enumerate(results)
  if isa(v, BatchProcessingError)
@@ -190,13 +190,11 @@ function process_batch_errors!(p, f, results, on_error, retry_on, retry_n, retry
  if length(reprocess) > 0
  errors = [x[2] for x in reprocess]
  exceptions = [x.ex for x in errors]
- if (retry_n > 0) && all([retry_on(ex) for ex in exceptions])
- retry_n = retry_n - 1
+ state = start(retry_delays)
+ if (length(retry_delays) > 0) &&
+ (retry_check==nothing || all([retry_check(state,ex)[2] for ex in exceptions]))
  error_processed = pmap(p, f, [x.data for x in errors];
- on_error=on_error,
- retry_on=retry_on,
- retry_n=retry_n,
- retry_max_delay=retry_max_delay)
+ on_error = on_error, retry_delays = collect(retry_delays)[2:end], retry_check = retry_check)
  elseif on_error !== nothing
  error_processed = map(on_error, exceptions)
  else

base/workerpool.jl

@@ -208,7 +208,7 @@ end
 """
  remote([::AbstractWorkerPool], f) -> Function
 
-Returns a lambda that executes function `f` on an available worker
+Returns an anonymous function that executes function `f` on an available worker
 using [`remotecall_fetch`](@ref).
 """
 remote(f) = (args...; kwargs...)->remotecall_fetch(f, default_worker_pool(), args...; kwargs...)

doc/src/stdlib/base.md

@@ -231,6 +231,7 @@ Core.UndefRefError
 Core.UndefVarError
 Base.InitError
 Base.retry
+Base.ExponentialBackOff
 ```
 
 ## Events

test/error.jl

@@ -1,5 +1,14 @@
 # This file is a part of Julia. License is MIT: http:https://julialang.org/license
 
+@test length(ExponentialBackOff(n=10)) == 10
+@test collect(ExponentialBackOff(n=10, first_delay=0.01))[1] == 0.01
+@test maximum(ExponentialBackOff(n=10, max_delay=0.06)) == 0.06
+ratio(x) = x[2:end]./x[1:end-1]
+@test all(x->isapprox(x,10.0), ratio(collect(
+ ExponentialBackOff(n=10, max_delay=Inf, factor=10, jitter=0.0))))
+srand(12345)
+@test (mean(ratio(collect(ExponentialBackOff(n=100, max_delay=Inf, factor=1, jitter=0.1)))) - 1.0) < 1e-4
+
 let
  function foo_error(c, n)
  c[1] += 1
@@ -16,42 +25,42 @@ let
 
  # Success on second attempt
  c = [0]
- @test retry(foo_error;n=1)(c,1) == 7
+ @test retry(foo_error)(c,1) == 7
  @test c[1] == 2
 
  # 2 failed retry attempts, so exception is raised
  c = [0]
- ex = try retry(foo_error;n=2)(c,3) catch e; e end
+ ex = try retry(foo_error, delays=ExponentialBackOff(n=2))(c,3) catch e; e end
  @test ex.msg == "foo"
  @test c[1] == 3
 
  c = [0]
- ex = try retry(foo_error, ErrorException)(c,2) catch e; e end
+ ex = try retry(foo_error, check=(s,e)->(s,isa(e, ErrorException)))(c,2) catch e; e end
  @test typeof(ex) == ErrorException
  @test ex.msg == "foo"
  @test c[1] == 2
 
  c = [0]
- ex = try retry(foo_error, e->e.msg == "foo")(c,2) catch e; e end
+ ex = try retry(foo_error, check=(s,e)->(s,e.msg == "foo"))(c,2) catch e; e end
  @test typeof(ex) == ErrorException
  @test ex.msg == "foo"
  @test c[1] == 2
 
  # No retry if condition does not match
  c = [0]
- ex = try retry(foo_error, e->e.msg == "bar"; n=3)(c,2) catch e; e end
+ ex = try retry(foo_error, check=(s,e)->(s,e.msg == "bar"))(c,2) catch e; e end
  @test typeof(ex) == ErrorException
  @test ex.msg == "foo"
  @test c[1] == 1
 
  c = [0]
- ex = try retry(foo_error, e->e.http_status_code == "503")(c,2) catch e; e end
+ ex = try retry(foo_error, check=(s,e)->(s,try e.http_status_code == "503" end != true))(c,2) catch e; e end
  @test typeof(ex) == ErrorException
  @test ex.msg == "foo"
- @test c[1] == 1
+ @test c[1] == 2
 
  c = [0]
- ex = try retry(foo_error, SystemError)(c,2) catch e; e end
+ ex = try retry(foo_error, check=(s,e)->(s,isa(e,SystemError)))(c,2) catch e; e end
  @test typeof(ex) == ErrorException
  @test ex.msg == "foo"
  @test c[1] == 1

test/parallel_exec.jl

@@ -683,9 +683,8 @@ pmap_args = [
  (:distributed, [:default, false]),
  (:batch_size, [:default,2]),
  (:on_error, [:default, e -> unmangle_exception(e).msg == "foobar"]),
- (:retry_on, [:default, e -> unmangle_exception(e).msg == "foobar"]),
- (:retry_n, [:default, typemax(Int)-1]),
- (:retry_max_delay, [0, 0.001])
+ (:retry_delays, [:default, fill(0.01, 1000)]),
+ (:retry_check, [:default, (s,e) -> (s,unmangle_exception(e).msg == "foobar")]),
  ]
 
 kwdict = Dict()
@@ -702,7 +701,7 @@ function walk_args(i)
 
  testw = kwdict[:distributed] === false ? [1] : workers()
 
- if (kwdict[:on_error] === :default) && (kwdict[:retry_n] === :default)
+ if (kwdict[:on_error] === :default) && (kwdict[:retry_delays] === :default)
  mapf = x -> (x*2, myid())
  results_test = pmap_res -> begin
  results = [x[1] for x in pmap_res]
@@ -712,7 +711,7 @@ function walk_args(i)
  @test p in pids
  end
  end
- elseif kwdict[:retry_n] !== :default
+ elseif kwdict[:retry_delays] !== :default
  mapf = x -> iseven(myid()) ? error("foobar") : (x*2, myid())
  results_test = pmap_res -> begin
  results = [x[1] for x in pmap_res]
@@ -726,7 +725,7 @@ function walk_args(i)
  end
  end
  end
- else (kwdict[:on_error] !== :default) && (kwdict[:retry_n] === :default)
+ else (kwdict[:on_error] !== :default) && (kwdict[:retry_delays] === :default)
  mapf = x -> iseven(x) ? error("foobar") : (x*2, myid())
  results_test = pmap_res -> begin
  w = testw