evolution.tcc

Go to the documentation of this file.

 
#if !defined(ULTRA_EVOLUTION_H)
#  error "Don't include this file directly, include the specific .h instead"
#endif
 
#if !defined(ULTRA_EVOLUTION_TCC)
#define      ULTRA_EVOLUTION_TCC
 
namespace internal
{
 
enum class phase {evolution, refinement};
 
struct print_status
{
  phase run_phase {phase::evolution};
 
  std::atomic<unsigned> planned_steps {0};
  std::atomic<unsigned> completed_steps {0};
 
  timer from_start {};
  timer from_last_msg {};
 
  std::size_t helpers {0};
 
  void new_generation()
  {
    run_phase = phase::evolution;
    planned_steps.store(0, std::memory_order_relaxed);
    completed_steps.store(0, std::memory_order_relaxed);
  }
};
 
class refinement_schedule
{
public:
  explicit refinement_schedule(const parameters::refinement_parameters &rp)
    : rp_(rp)
  {
  }
 
  [[nodiscard]] bool ready(unsigned stagnation) const noexcept
  {
    return cooldown_left_ == 0 && stagnation >= rp_.stagnation_threshold;
  }
 
  void after_refinement_attempt(bool performed) noexcept
  {
    if (performed)
      cooldown_left_ = rp_.cooldown;
  }
 
  void tick() noexcept
  {
    if (cooldown_left_)
      --cooldown_left_;
  }
 
private:
  const parameters::refinement_parameters &rp_;
 
  unsigned cooldown_left_ {};
};

class task_group
{
public:
  explicit task_group(std::size_t workers) : pool_(workers)
  {
    Expects(workers);
  }

  template<class F, class... Args>
  requires std::same_as<std::invoke_result_t<F, Args...>, void>
  void submit(F &&f, Args&&... args)
  {
    tasks_.push_back(pool_.submit(std::forward<F>(f),
                                  std::forward<Args>(args)...));
  }

  void wait()
  {
    pool_.wait();
    rethrow_exceptions();
  }

  template<class Rep, class Period>
  [[nodiscard]] bool wait_for(const std::chrono::duration<Rep, Period> &timeout)
  {
    if (!pool_.wait_for(timeout))
      return false;
 
    rethrow_exceptions();
    return true;
  }
 
private:
  void rethrow_exceptions()
  {
    std::exception_ptr first_exception;
 
    for (auto &task : tasks_)
      try
      {
        task.get();
      }
      catch (...)
      {
        if (!first_exception)
          first_exception = std::current_exception();
      }
 
    tasks_.clear();
 
    if (first_exception)
      std::rethrow_exception(first_exception);
  }
 
  ultra::thread_pool pool_;
  std::vector<std::future<void>> tasks_ {};
};

template<Evaluator E>
[[nodiscard]] std::size_t set_evaluator_parallelism(
  std::size_t allocated, std::size_t min_samples_per_eval_thread, E &eva)
{
  if constexpr (requires { eva.set_parallel_evaluation(0, 0); })
  {
    const auto capacity(hardware_threads());
    const auto helpers(allocated && allocated < capacity ?
                       (capacity - allocated) / allocated : 0);
 
    ultraDEBUG("Configure evaluator's parallelism to {}", helpers);
    if (eva.set_parallel_evaluation(helpers, min_samples_per_eval_thread))
      return helpers;
  }
 
  return 0;
}
 
}  // namespace internal

template<Evaluator E>
evolution<E>::evolution(const problem &prob, E &eva) : pop_(prob), eva_(eva)
{
  Ensures(is_valid());
 
  ultraDEBUG("Creating a new instance of evolution class");
}

template<Evaluator E>
bool evolution<E>::stop_condition() const
{
  const auto planned_generations(pop_.problem().params.evolution.generations);
  Expects(planned_generations);
 
  // Check the number of generations.
  if (sum_.generation >= planned_generations)
    return true;
 
  if (console.user_stop())
    return true;
 
  return external_stop_source_.stop_requested();
}
 
template<Evaluator E>
void evolution<E>::print(message m, internal::print_status &ps) const
{
  if (!emit_messages_ || log::reporting_level > log::lPAROUT)
    return;
 
  const std::string tags(tag_.empty() ? tag_ : "[" + tag_ + "] ");
 
  const std::string_view str_phase(ps.run_phase == internal::phase::evolution
                                   ? "evolution" : "refinement");
 
  const auto elapsed(ps.from_start.elapsed());
 
  if (m == message::summary)
  {
    const auto fitness(internal::streamed(sum_.best().fit));
 
    ultraPAROUT("{}{:>8}{:>8}{:>12}:{:>13}", tags,
                lexical_cast<std::string>(elapsed), sum_.generation,
                str_phase, fitness);
  }
  else  // message::status
  {
    std::string perc;
 
    const auto planned(ps.planned_steps.load(std::memory_order_relaxed));
    if (planned)
    {
      const auto completed(ps.completed_steps.load(std::memory_order_relaxed));
      perc = std::to_string(std::uintmax_t(100) * completed / planned) + "%";
    }
 
    std::string line;
    if (log::reporting_level == log::lPAROUT)
    {
      static constexpr std::string_view chrs("|/-\\");
 
      line = std::format("{}{} {} {}",
                         tags,
                         chrs[elapsed.count() % chrs.size()],
                         str_phase,
                         perc);
    }
    else if (log::reporting_level <= log::lSTDOUT)
    {
      const auto seconds(
        std::max(std::chrono::duration_cast<std::chrono::seconds>(elapsed),
                 1s).count());
 
      line = std::format("{}  gen {} {} {}",
                         lexical_cast<std::string>(elapsed),
                         sum_.generation,
                         str_phase,
                         perc);
 
      if (sum_.generation)
      {
        const double gph(3600.0 * sum_.generation / seconds);
        line += std::format("  [{}L", pop_.layers());
        if (ps.helpers)
          line += std::format(" x {}T", 1 + ps.helpers);
        line += std::format(" | {0:.{1}f}gph]", gph, gph > 2.0 ? 0 : 1);
      }
    }
 
    console.status_line(line);
  }
 
  if (ps.run_phase == internal::phase::evolution)
    ps.from_last_msg.restart();
}

template<Evaluator E>
evolution<E> &evolution<E>::logger(search_log &sl)
{
  search_log_ = &sl;
  return *this;
}
 
template<Evaluator E>
evolution<E> &evolution<E>::refinement(refinement_callback_t f)
{
  refinement_callback_ = std::move(f);
  return *this;
}
 
template<Evaluator E>
evolution<E> &evolution<E>::messages(bool m) noexcept
{
  emit_messages_ = m;
  return *this;
}

template<Evaluator E>
evolution<E> &evolution<E>::tag(const std::string &t)
{
  tag_ = t;
  return *this;
}

template<Evaluator E>
evolution<E> &evolution<E>::shake_function(shake_function_callback_t f)
{
  shake_ = std::move(f);
  return *this;
}

template<Evaluator E>
evolution<E> &evolution<E>::after_generation(after_generation_callback_t f)
{
  after_generation_callback_ = std::move(f);
  return *this;
}
 
template<Evaluator E>
evolution<E> &evolution<E>::on_new_best(on_new_best_callback_t f)
{
  sum_.on_new_best(std::move(f));
  return *this;
}

template<Evaluator E>
evolution<E> &evolution<E>::stop_source(std::stop_source ss)
{
  external_stop_source_ = ss;
  return *this;
}

template<Evaluator E>
template<Population P>
auto evolution<E>::refinement_tournament(const P &ref_pop)
{
  const auto &params(pop_.problem().params);
  const auto t_size(params.refinement.tournament_size
                    ? params.refinement.tournament_size
                    : params.evolution.tournament_size);
  assert(t_size > 0);
  assert(t_size <= parameters::evolution_parameters::max_tournament_size);
 
  auto best_coord(random::coord(ref_pop));
  auto best_fit(eva_(ref_pop[best_coord]));
 
  for (std::size_t i(1); i < t_size; ++i)
  {
    const auto c(random::coord(ref_pop));
    const auto fit(eva_(ref_pop[c]));
 
    if (fit > best_fit)
    {
      best_coord = c;
      best_fit = fit;
    }
  }
 
  return best_coord;
}

template<Evaluator E>
template<Population P>
bool evolution<E>::perform_refinement(P &ref_pop, internal::print_status &ps)
{
  if (!ref_pop.size())
    return false;
 
  const auto refinement_fraction(pop_.problem().params.refinement.fraction);
  Expects(in_0_1(refinement_fraction));
 
  if (issmall(refinement_fraction))
    return false;
 
  // ********* Select refinement candidates *********
  const auto base_n(
    static_cast<std::size_t>(refinement_fraction * ref_pop.size()));
  const std::size_t n(std::max(base_n, 1uz));
 
  struct refinement_candidate
  {
    typename P::coord coord;
    individual_t initial;
    scored_individual<individual_t, fitness_t> refined;
  };
 
  std::vector<refinement_candidate> candidates;
  candidates.reserve(n);
 
  std::set<typename P::coord> seen;
 
  for (std::size_t i(0); i < n; ++i)
  {
    const auto coord(refinement_tournament(ref_pop));
 
    if (seen.insert(coord).second)
      candidates.push_back({coord, ref_pop[coord], {}});
  }
 
  if (candidates.empty())
    return false;
 
  // ********* Refine candidates in parallel *********
  ps.run_phase = internal::phase::refinement;
  ps.planned_steps.store(static_cast<unsigned>(candidates.size()),
                         std::memory_order_relaxed);
  ps.completed_steps.store(0, std::memory_order_relaxed);
 
  const auto threads(std::min(candidates.size(), hardware_threads()));
 
  ps.helpers = internal::set_evaluator_parallelism(
    threads, pop_.problem().params.min_samples_per_eval_thread, eva_);
 
  internal::task_group tasks(threads);
 
  for (std::size_t i(0); i < candidates.size(); ++i)
    tasks.submit(
      [&, i]
      {
        // Suppress backend output here: the evolution loop reports refinement
        // progress and multiple backend instances may run concurrently.
        refiner optimiser(pop_.problem(), false);
 
        auto ind(candidates[i].initial);
 
        const auto fit(
          optimiser.optimise(ind, eva_, refinement_callback_));
 
        if (fit)
          candidates[i].refined = scored_individual(ind, *fit);
 
        ps.completed_steps.fetch_add(1, std::memory_order_relaxed);
        print(message::status, ps);
      });
 
  tasks.wait();
 
  // Commit refinement results.
  for (const auto &result : candidates)
    if (!result.refined.empty())
    {
      if (sum_.update_if_better(result.refined))
        print(message::summary, ps);
 
      ref_pop[result.coord] = result.refined.ind;
    }
 
  // Restore evolution phase and population-layer parallelism.
  ps.run_phase = internal::phase::evolution;
  ps.helpers = internal::set_evaluator_parallelism(
    pop_.layers(), pop_.problem().params.min_samples_per_eval_thread,
    eva_);
 
  return true;
}

template<Evaluator E>
template<template<class> class ES>
summary<typename evolution<E>::individual_t,
        typename evolution<E>::fitness_t> evolution<E>::run()
{
  Expects(sum_.generation == 0);
 
  const auto &params(pop_.problem().params);
 
  if (params.needs_init())
    throw std::logic_error("Parameters still contain auto-tune values. "
                           "Call parameters::init() (or use src::search) "
                           "before evolution::run().");
 
  using namespace std::chrono_literals;
 
  std::stop_source source;
 
  // An external stop request immediately requests cancellation on the local
  // source.
  std::stop_callback external_stop_callback(
    external_stop_source_.get_token(),
    [&source] { source.request_stop(); });
 
  if (source.stop_requested())
    return sum_;
 
  ES<E> strategy(pop_.problem(), eva_);
 
  // Last observed best (for progress reporting).
  scored_individual progress_best(sum_.best());
 
  internal::print_status ps;
  internal::refinement_schedule ref_sched(params.refinement);
 
  // Asynchronous population update: each newly generated offspring can replace
  // an individual of the current population (aka steady state population).
  // Asynchronous update permits new individual to contribute to the evolution
  // immediately and can speed up the convergence.
  const auto evolve_subpop(
    [&, stop_token = source.get_token()](auto subpop_it)
    {
      auto evolve(strategy.operations(pop_, subpop_it,
                                      sum_.starting_status()));
 
      // We must use `safe_size()` because other threads might migrate
      // individuals in this subpopulation.
      auto cycles(subpop_it->safe_size());
 
      ps.planned_steps.fetch_add(cycles, std::memory_order_relaxed);
 
      while (cycles--)
        if (!stop_token.stop_requested())
        {
          evolve();
          ps.completed_steps.fetch_add(1, std::memory_order_relaxed);
        }
    });
 
  const auto print_and_update_if_better(
    [&](auto candidate)
    {
      if (progress_best < candidate)
      {
        progress_best = candidate;
        print(message::summary, ps);
        return true;
      }
 
      return false;
    });
 
  ultraDEBUG("Calling evolution_strategy init method");
  strategy.init(pop_);  // strategy-specific customisation point
 
  internal::task_group tasks(strategy.max_parallelism(pop_));
 
  while (!source.stop_requested() && !stop_condition())
  {
    ps.new_generation();
 
    if (shake_ && shake_(sum_.generation) == evaluation_context::changed)
    {
      invalidate_cache_if_supported(eva_);
 
      if (const auto best(sum_.best()); !best.empty())
      {
        sum_.set_best_fitness(eva_(best.ind));
        progress_best = sum_.best();
      }
    }
 
    ps.helpers = internal::set_evaluator_parallelism(
      pop_.layers(), pop_.problem().params.min_samples_per_eval_thread, eva_);
 
    ultraDEBUG("Launching tasks for generation {}", sum_.generation);
    const auto subpops(pop_.range_of_layers());
 
    for (auto l(subpops.begin()); l != subpops.end(); ++l)
      tasks.submit(evolve_subpop, l);
 
    ultraDEBUG("Tasks running");
 
    while (!tasks.wait_for(10ms))
    {
      if (ps.from_last_msg.elapsed() > 2s
          && !print_and_update_if_better(sum_.best()))
        print(message::status, ps);
 
      if (!source.stop_requested() && stop_condition()) [[unlikely]]
      {
        source.request_stop();
        ultraDEBUG("Sending closing message to tasks");
      }
    }
 
    print_and_update_if_better(sum_.best());
 
    if (source.stop_requested())
      break;
 
    if constexpr (RefinementStrategy<ES<E>, decltype(pop_)>)
    {
      if (refinement_callback_)
      {
        if (ref_sched.ready(sum_.stagnation()))
        {
          bool performed(false);
 
          if (auto *ref_pop = strategy.refinement_subgroup(pop_))
            performed = perform_refinement(*ref_pop, ps);
 
          ref_sched.after_refinement_attempt(performed);
 
          // `performed` only means refinement ran. It may still have updated
          // and reported `sum_`, so refresh the local progress cache.
          if (performed)
            progress_best = sum_.best();
        }
        else
          ref_sched.tick();
      }
    }
 
    sum_.az = analyzer(pop_, eva_);
    if (search_log_)
      search_log_->save_snapshot(pop_, sum_);
 
    strategy.after_generation(pop_, sum_);  // strategy-specific bookkeeping
    if (after_generation_callback_)
      after_generation_callback_(pop_, sum_);
 
    ++sum_.generation;
  }
 
  sum_.elapsed = ps.from_start.elapsed();
 
  if (emit_messages_)
    ultraINFO("Evolution completed at generation: {}. Elapsed time: {}",
                sum_.generation,
                lexical_cast<std::string>(ps.from_start.elapsed()));
 
  return sum_;
}

template<Evaluator E>
bool evolution<E>::is_valid() const
{
  return pop_.is_valid();
}
 
#endif  // include guard