<commit_before>/*
 * Copyright (C) 2009, <NAME>.
 *
 * This software is provided "as is" without warranty of any kind,
 * either expressed or implied, including but not limited to the
 * implied warranties of fitness for a particular purpose.
 *
 * See the LICENSE file for more information.
 */
#include <libport/cstdlib>
#include <iostream>

#include <libport/bind.hh>
#include <boost/lambda/lambda.hpp>
#include <boost/mem_fn.hpp>

#include <libport/compiler.hh>     // For LIBPORT_DEBUG
#include <libport/containers.hh>
#include <libport/debug.hh>
#include <libport/foreach.hh>
#include <libport/indent.hh>
#include <libport/separate.hh>

#include <sched/job.hh>

GD_ADD_CATEGORY(sched);

namespace sched
{
  StopException::StopException(unsigned depth, boost::any payload)
    : depth_(depth)
    , payload_(payload)
  {
  }

  /*-----------------.
  | Job::Collector.  |
  `-----------------*/

  Job::Collector::~Collector()
  {
    foreach (rJob& child, *this)
      parent_->terminate_child(child);
  }

  void
  Job::Collector::collect()
  {
    for (iterator i = begin(); i != end(); /* nothing. */)
      if ((*i)->terminated())
        {
          parent_->terminate_child(*i);
          i = erase(i);
        }
      else
        ++i;
  }

  std::ostream&
  Job::Collector::dump(std::ostream& o) const
  {
    return o << "{" << libport::incendl
             // Explicit parameters required by MSWC 2005.
             << libport::separate<const Collector,
                                  std::ostream&(*)(std::ostream&)>
                                  (*this, libport::iendl)
             << libport::decendl << "}";
  }

  std::ostream&
  operator<< (std::ostream& o, const Job::Collector& c)
  {
    return c.dump(o);
  }


  /*------.
  | Job.  |
  `------*/

  unsigned int Job::alive_jobs_ = 0;

  std::ostream&
  Job::dump(std::ostream& o) const
  {
    o << "Job(" << name_ << ")"
      << " "
      << (child_job() ? "child job" : "root job")
      << " state: " << state_;
    if (!children_.empty())
      o << " Children:" << libport::incendl
        << "{" << libport::incendl
        << children_ << libport::decendl
        << "}" << libport::decindent;
    return o;
  }

  void
  Job::run()
  {
    GD_CATEGORY(sched);
    assert(state_ == to_start);
    LIBPORT_DEBUG("In Job::run for " << this);

    // We may get interrupted during our first run, in which case
    // we better not be in the to_start state while we are executing
    // or we would get removed abruptly from the scheduler pending_
    // list.
    state_ = running;
    try {
      if (has_pending_exception() &&
	  dynamic_cast<SchedulerException*>(pending_exception_.get()))
	check_for_pending_exception();
      work();
    }
    catch (TerminateException&)
    {
      // Normal termination requested.
    }
    catch (StopException&)
    {
      // Termination through "stop" or "block" on a top-level tag,
      // that is a tag inherited at the job creation time.
    }
    catch (const exception& e)
    {
      // Rethrow the exception into the parent job if it exists.
      if (parent_)
	parent_->async_throw(ChildException(e.clone()));
    }
    catch (const std::exception& e)
    {
      // Exception is lost and cannot be propagated properly but can
      // be printed onto the console for diagnostic purpose.
      GD_FERROR("job %s: Exception `%s' caught, losing it",
                (this)(e.what()));
    }
    catch (...)
    {
      // Exception is lost and cannot be propagated properly.
      GD_FERROR("job %s: Unknown exception caught, losing it",
                (this));
    }

    terminate_cleanup();

    // We should never go there as the scheduler will have terminated us.
    pabort("scheduler did not terminate us");
  }

  void
  Job::terminate_now()
  {
    // We have to terminate our children as well.
    terminate_jobs(children_);
    terminate_asap();
  }

  void
  Job::terminate_asap()
  {<commit_msg>sched: Fix a memory leak for each terminating coroutine.<commit_after>/*
 * Copyright (C) 2009, <NAME>.
 *
 * This software is provided "as is" without warranty of any kind,
 * either expressed or implied, including but not limited to the
 * implied warranties of fitness for a particular purpose.
 *
 * See the LICENSE file for more information.
 */
#include <libport/cstdlib>
#include <iostream>

#include <libport/bind.hh>
#include <boost/lambda/lambda.hpp>
#include <boost/mem_fn.hpp>

#include <libport/compiler.hh>     // For LIBPORT_DEBUG
#include <libport/containers.hh>
#include <libport/debug.hh>
#include <libport/foreach.hh>
#include <libport/indent.hh>
#include <libport/separate.hh>

#include <sched/job.hh>

GD_ADD_CATEGORY(sched);

namespace sched
{
  StopException::StopException(unsigned depth, boost::any payload)
    : depth_(depth)
    , payload_(payload)
  {
  }

  /*-----------------.
  | Job::Collector.  |
  `-----------------*/

  Job::Collector::~Collector()
  {
    foreach (rJob& child, *this)
      parent_->terminate_child(child);
  }

  void
  Job::Collector::collect()
  {
    for (iterator i = begin(); i != end(); /* nothing. */)
      if ((*i)->terminated())
        {
          parent_->terminate_child(*i);
          i = erase(i);
        }
      else
        ++i;
  }

  std::ostream&
  Job::Collector::dump(std::ostream& o) const
  {
    return o << "{" << libport::incendl
             // Explicit parameters required by MSWC 2005.
             << libport::separate<const Collector,
                                  std::ostream&(*)(std::ostream&)>
                                  (*this, libport::iendl)
             << libport::decendl << "}";
  }

  std::ostream&
  operator<< (std::ostream& o, const Job::Collector& c)
  {
    return c.dump(o);
  }


  /*------.
  | Job.  |
  `------*/

  unsigned int Job::alive_jobs_ = 0;

  std::ostream&
  Job::dump(std::ostream& o) const
  {
    o << "Job(" << name_ << ")"
      << " "
      << (child_job() ? "child job" : "root job")
      << " state: " << state_;
    if (!children_.empty())
      o << " Children:" << libport::incendl
        << "{" << libport::incendl
        << children_ << libport::decendl
        << "}" << libport::decindent;
    return o;
  }

  void
  Job::run()
  {
    // This function never terminates properly: it yields and is never
    // called back. So perform RAII in a subblock to ensure that GD_CATEGORY
    // releases its memory.
    {
      GD_CATEGORY(sched);
      assert(state_ == to_start);
      LIBPORT_DEBUG("In Job::run for " << this);

      // We may get interrupted during our first run, in which case
      // we better not be in the to_start state while we are executing
      // or we would get removed abruptly from the scheduler pending_
      // list.
      state_ = running;
      try {
        if (has_pending_exception() &&
            dynamic_cast<SchedulerException*>(pending_exception_.get()))
	check_for_pending_exception();
        work();
      }
      catch (TerminateException&)
      {
      // Normal termination requested.
      }
      catch (StopException&)
      {
        // Termination through "stop" or "block" on a top-level tag,
        // that is a tag inherited at the job creation time.
      }
      catch (const exception& e)
      {
        // Rethrow the exception into the parent job if it exists.
        if (parent_)
          parent_->async_throw(ChildException(e.clone()));
      }
      catch (const std::exception& e)
      {
        GD_FERROR("job %s: Exception `%s' caught, losing it",
                  (this)(e.what()));
      }
      catch (...)
      {
        // Exception is lost and cannot be propagated properly.
        GD_FERROR("job %s: Unknown exception caught, losing it",
                  (this));
      }
    }

    terminate_cleanup();

    // We should never go there as the scheduler will have terminated us.
    pabort("scheduler did not terminate us");
  }

  void
  Job::terminate_now()
  {
    // We have to terminate our children as well.
    terminate_jobs(children_);
    terminate_asap();
  }

  void
  Job::terminate_asap()
  {
{'fraction_non_alphanumeric': 0.1480291610033362, 'fraction_numerical': 0.0022241443222537995, 'mean_word_length': 2.2782503037667072, 'pattern_counts': {'":': 0, '<': 121, '<?xml version=': 0, '>': 41, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3049420', 'n_tokens_mistral': 2553, 'n_tokens_neox': 2469, 'n_words': 678}