Parallelize solve_problems

src/bmark_solvers.jl

@@ -10,16 +10,22 @@ Run a set of solvers on a set of problems.
 * other positional arguments accepted by `solve_problems`, except for a solver name
 
 #### Keyword arguments
-Any keyword argument accepted by `solve_problems`
+* `parallel::Bool`: whether to run the problems in parallel (default: false). The user is responsible for ensuring that the solvers and problems are thread-safe if this is set to true. Moreover, the number of Julia threads must be set to a value greater than 1 (see `JULIA_NUM_THREADS` environment variable).
+All other keyword arguments are given to `solve_problems` or `solve_problems_parallel` if `parallel` is true.
 
 #### Return value
 A Dict{Symbol, AbstractExecutionStats} of statistics.
 """
-function bmark_solvers(solvers::Dict{Symbol, <:Any}, args...; kwargs...)
+function bmark_solvers(solvers::Dict{Symbol, <:Any}, args...; parallel::Bool = false, kwargs...)
   stats = Dict{Symbol, DataFrame}()
   for (name, solver) in solvers
     @info "running solver $name"
-    stats[name] = solve_problems(solver, name, args...; kwargs...)
+    if parallel && Threads.nthreads() > 1
+      stats[name] = solve_problems_parallel(solver, name, args...; kwargs...)
+    else
+      parallel && @warn "SolverBenchmarks.jl: parallel is set to true but the number of threads is $(Threads.nthreads()). Running in serial mode."
+      stats[name] = solve_problems(solver, name, args...; kwargs...)
+    end
   end
   return stats
 end

src/run_solver.jl

@@ -192,3 +192,289 @@ function solve_problems(
   end
   return stats
 end
+
+"""
+    solve_problems_parallel(solver, solver_name, problems; kwargs...)
+
+Apply a solver to a set of problems in parallel.
+
+#### Arguments
+* `solver`: the function name of a solver;
+* `solver_name`: name of the solver;
+* `problems`: the set of problems to pass to the solver, as an iterable of
+  `AbstractNLPModel`. It is recommended to use a generator expression (necessary for
+  CUTEst problems).
+
+#### Keyword arguments
+* `solver_logger::AbstractLogger`: logger wrapping the solver call (default: `NullLogger`);
+* `reset_problem::Bool`: reset the problem's counters before solving (default: `true`);
+* `skipif::Function`: function to be applied to a problem and return whether to skip it
+  (default: `x->false`);
+* `colstats::Vector{Symbol}`: summary statistics for the logger to output during the
+benchmark (default: `[:name, :nvar, :ncon, :status, :elapsed_time, :objective, :dual_feas, :primal_feas]`), solver's `solver_specific` scalar entries are appended as extra columns;
+* `info_hdr_override::Dict{Symbol,String}`: header overrides for the summary statistics
+  (default: use default headers);
+* `prune`: do not include skipped problems in the final statistics (default: `true`);
+* any other keyword argument to be passed to the solver.
+
+#### Return value
+* a `DataFrame` where each row is a problem, minus the skipped ones if `prune` is true.
+
+#### Warning
+This function assumes that the solver and the problems are thread-safe. If it is not, switch to the serial version (see `solve_problems`).
+"""
+function solve_problems_parallel(
+  solver,
+  solver_name::TName,
+  problems;
+  solver_logger::AbstractLogger = NullLogger(),
+  reset_problem::Bool = true,
+  skipif::Function = x -> false,
+  colstats::Vector{Symbol} = [
+    :solver_name,
+    :name,
+    :nvar,
+    :ncon,
+    :status,
+    :iter,
+    :elapsed_time,
+    :objective,
+    :dual_feas,
+    :primal_feas,
+  ],
+  info_hdr_override::Dict{Symbol, String} = Dict{Symbol, String}(:solver_name => "Solver"),
+  prune::Bool = true,
+  kwargs...,
+) where {TName}
+  f_counters = collect(fieldnames(Counters))
+  fnls_counters = collect(fieldnames(NLSCounters))[2:end] # Excludes :counters
+  ncounters = length(f_counters) + length(fnls_counters)
+  types = [
+    TName
+    Int
+    String
+    Int
+    Int
+    Int
+    Symbol
+    Float64
+    Float64
+    Int
+    Float64
+    Float64
+    fill(Int, ncounters)
+    String
+  ]
+  names = [
+    :solver_name
+    :id
+    :name
+    :nvar
+    :ncon
+    :nequ
+    :status
+    :objective
+    :elapsed_time
+    :iter
+    :dual_feas
+    :primal_feas
+    f_counters
+    fnls_counters
+    :extrainfo
+  ]
+  stats = DataFrame(names .=> [T[] for T in types])
+  stats_lock = ReentrantLock()
+
+  specific = Symbol[]
+
+  col_idx = indexin(colstats, names)
+
+  first_problem = true
+  nb_unsuccessful_since_start = 0
+  @info log_header(colstats, types[col_idx], hdr_override = info_hdr_override)
+
+  # Make a first serial run until first_problem is false
+  final_id = 0
+  for (id, problem) in enumerate(problems)
+    first_problem, nb_unsuccessful_since_start = _locked_run_problem(
+      id,
+      problem,
+      stats,
+      solver,
+      solver_name,
+      solver_logger,
+      skipif,
+      stats_lock,
+      first_problem,
+      nb_unsuccessful_since_start,
+      specific,
+      prune,
+      reset_problem,
+      ncounters,
+      f_counters,
+      fnls_counters,
+      col_idx;
+      kwargs...,
+    )
+    final_id = id
+    !first_problem && break
+  end
+
+  # Remaining problems to run in parallel
+  rem_problems = Iterators.drop(problems, final_id)
+
+  # Start parallel run
+  Threads.@sync for (offset, problem) in enumerate(rem_problems)
+    id = final_id + offset
+    Threads.@spawn begin
+      _locked_run_problem(
+        id,
+        problem,
+        stats,
+        solver,
+        solver_name,
+        solver_logger,
+        skipif,
+        stats_lock,
+        first_problem,
+        nb_unsuccessful_since_start,
+        specific,
+        prune,
+        reset_problem,
+        ncounters,
+        f_counters,
+        fnls_counters,
+        col_idx;
+        kwargs...,
+      )
+    end
+  end
+  return stats
+end
+
+# Run a problem and lock the stats DataFrame for parallel write.
+function _locked_run_problem(
+  id::Int,
+  problem,
+  stats::DataFrame,
+  solver, 
+  solver_name::Symbol,
+  solver_logger::AbstractLogger,
+  skipif::Function,
+  stats_lock::ReentrantLock,
+  first_problem::Bool,
+  nb_unsuccessful_since_start::Int,
+  specific::Vector{Symbol},
+  prune::Bool,
+  reset_problem::Bool,
+  ncounters::Int,
+  f_counters,
+  fnls_counters,
+  col_idx::AbstractVector;
+  kwargs...
+)
+  if reset_problem
+    NLPModels.reset!(problem)
+  end
+  nequ = problem isa AbstractNLSModel ? problem.nls_meta.nequ : 0
+  problem_info = [id; get_name(problem); get_nvar(problem); get_ncon(problem); nequ]
+  skipthis = skipif(problem)
+
+  if skipthis
+    if first_problem && !prune
+      nb_unsuccessful_since_start += 1
+    end
+    prune || lock(stats_lock) do 
+      push!(
+        stats,
+        [
+          solver_name
+          problem_info
+          :exception
+          Inf
+          Inf
+          0
+          Inf
+          Inf
+          fill(0, ncounters)
+          "skipped"
+          fill(missing, length(specific))
+        ],
+      )
+      @info log_row(stats[end, col_idx])
+    end
+    finalize(problem)
+  else
+    try
+      s = with_logger(solver_logger) do
+        solver(problem; kwargs...)
+      end
+      if first_problem
+        for (k, v) in s.solver_specific
+          if !(typeof(v) <: AbstractVector)
+            insertcols!(
+              stats,
+              ncol(stats) + 1,
+              k => Vector{Union{typeof(v), Missing}}(undef, nb_unsuccessful_since_start),
+            )
+            push!(specific, k)
+          end
+        end
+        first_problem = false
+      end
+      counters_list = [getfield(NLPModels, f)(problem) for f in f_counters]
+        nls_counters_list =
+          problem isa AbstractNLSModel ? [getfield(problem.counters, f) for f in fnls_counters] :
+          zeros(Int, length(fnls_counters))
+
+        lock(stats_lock) do 
+          push!(
+            stats,
+            [
+              solver_name
+              problem_info
+              s.status
+              s.objective
+              s.elapsed_time
+              s.iter
+              s.dual_feas
+              s.primal_feas
+              counters_list
+              nls_counters_list
+              ""
+              [s.solver_specific[k] for k in specific]
+            ],
+          )
+          @info log_row(stats[end, col_idx])
+        end
+    catch e
+      @error "caught exception" e
+      if first_problem
+        nb_unsuccessful_since_start += 1
+      end
+
+      lock(stats_lock) do 
+        push!(
+          stats,
+          [
+            solver_name
+            problem_info
+            :exception
+            Inf
+            Inf
+            0
+            Inf
+            Inf
+            fill(0, ncounters)
+            string(e)
+            fill(missing, length(specific))
+          ],
+        )
+        @info log_row(stats[end, col_idx])
+      end
+    finally
+      finalize(problem)
+    end
+  end
+  return first_problem, nb_unsuccessful_since_start
+end