<commit_before>#include <util/singleton.h>
#include <util/scheduler.h>
#include <util/timer.h>
#include <iostream>

#include <map>
#include <boost/thread.hpp>

using namespace std;

namespace izenelib{
namespace util{

class QueueTimer : public Timer
{
public:
    QueueTimer(void (*callback)(void *),
               void *arg,
               uint32_t due_time,
               uint32_t period)
            :callback_(callback),
             arg_(arg),
             due_time_(due_time),
             period_(period)
    {}

    bool start()
    {
        return Timer::start(due_time_, period_);
    }

    bool startNow()
    {
        return Timer::start(0, period_);
    }

    virtual void signaled()
    {
        callback_(arg_);
    }

private:
    void (*callback_)(void *);
    void *arg_;
    uint32_t due_time_;
    uint32_t period_;
};

struct ScheduleOP
{
    std::string name;
    uint32_t default_interval;
    uint32_t delay_start;
    boost::function<void (int)> callback;
    QueueTimer* timer;
    bool running;
    int calltype;
};

class SchedulerImpl
{
public:
    SchedulerImpl() {}

    virtual ~SchedulerImpl()
    {
        removeAllJobs();
    }

    void removeAllJobs()
    {
        boost::mutex::scoped_lock l(mutex_);
        for (std::map<std::string, ScheduleOP>::iterator itr = jobs_.begin();
                itr != jobs_.end(); ++itr)
        {
            ScheduleOP *job = &itr->second;
            if (job->timer)
            {
                job->timer->stop();
                delete job->timer;
            }
        }
        jobs_.clear();
    }

    bool addJob(const string &name, uint32_t default_interval,
                uint32_t delay_start, const boost::function<void (int)>& func)
    {
        boost::mutex::scoped_lock l(mutex_);
        std::map<string, ScheduleOP>::iterator find_itr = jobs_.find(name);
        if (find_itr != jobs_.end())
            return false;

        ScheduleOP newjob;
        newjob.name = name;
        newjob.calltype = 0;
        newjob.default_interval = default_interval;
        newjob.delay_start = delay_start;
        newjob.callback = func;
        newjob.timer = NULL;
        newjob.running = false;

        std::pair<std::map<std::string, ScheduleOP>::iterator, bool> result
        = jobs_.insert(make_pair(name, newjob));
        if (!result.second)
        {
            return false;
        }
        ScheduleOP *job = &result.first->second;

        uint32_t delay = job->delay_start;
        job->timer = new QueueTimer(&timerCallback, job, delay,
                                    job->default_interval);
        if (job->timer == NULL)
        {
            return false;
        }
        const bool started =  job->timer->start();
        if (started)
        {
            return true;
        }
        else
        {
            delete job->timer;
            return false;
        }
    }

    bool runJobImmediatly(const std::string& name, int calltype, bool sync)
    {
        boost::mutex::scoped_lock l(mutex_);
        std::map<std::string, ScheduleOP>::iterator itr = jobs_.find(name);
        if (itr == jobs_.end())
        {
 	    std::cout << "schedule job not found:" << name << std::endl;
            return false;
        }
        ScheduleOP *job = &itr->second;
        if (job && job->timer)
	{
	    job->calltype = calltype;
	    if (sync)
	    {
		if (job->running)
		{
		    std::cout << "schedule job already running:" << name << std::endl;
		    return false;
		}
		job->running = true;
		job->callback(calltype);
		job->running = false;
		job->calltype = 0;
		return true;
	    }
	    else
	    {
		return job->timer->startNow();
	    }
        }
	std::cout << "schedule job timer null:" << name << std::endl;
        return false;
    }

    bool removeJob(const std::string &name)
    {
        boost::mutex::scoped_lock l(mutex_);
        std::map<std::string, ScheduleOP>::iterator itr = jobs_.find(name);
        if (itr == jobs_.end())
        {
            return false;
        }
        else
        {
            ScheduleOP *job = &itr->second;
            if (job->timer != NULL)
            {
                job->timer->stop();
                delete job->timer;
            }
            jobs_.erase(itr);
            return true;
        }
    }

private:
    static void timerCallback(void *param)
    {
        ScheduleOP *job = reinterpret_cast<ScheduleOP *>(param);
        if (job->running)
            return;
        job->running = true;
        
        job->callback(job->calltype);
        job->running = false;
	job->calltype = 0;
    }

    std::map<std::string, ScheduleOP> jobs_;
    boost::mutex mutex_;
};

bool Scheduler::addJob(const string &name, uint32_t default_interval,
                       uint32_t delay_start, const boost::function<void (int)>& func)
{
    return Singleton<SchedulerImpl>::get()->addJob(name, default_interval, delay_start, func);
}

bool Scheduler::removeJob(const string &name)
{
    return Singleton<SchedulerImpl>::get()->removeJob(name);
}

void Scheduler::removeAllJobs()<commit_msg>cron job start immediatly in caller thread<commit_after>#include <util/singleton.h>
#include <util/scheduler.h>
#include <util/timer.h>
#include <iostream>

#include <map>
#include <boost/thread.hpp>

using namespace std;

namespace izenelib{
namespace util{

class QueueTimer : public Timer
{
public:
    QueueTimer(void (*callback)(void *),
               void *arg,
               uint32_t due_time,
               uint32_t period)
            :callback_(callback),
             arg_(arg),
             due_time_(due_time),
             period_(period)
    {}

    bool start()
    {
        return Timer::start(due_time_, period_);
    }

    bool startNow()
    {
        return Timer::start(0, period_);
    }

    virtual void signaled()
    {
        callback_(arg_);
    }

private:
    void (*callback_)(void *);
    void *arg_;
    uint32_t due_time_;
    uint32_t period_;
};

struct ScheduleOP
{
    std::string name;
    uint32_t default_interval;
    uint32_t delay_start;
    boost::function<void (int)> callback;
    QueueTimer* timer;
    volatile bool running;
};

class SchedulerImpl
{
public:
    SchedulerImpl() {}

    virtual ~SchedulerImpl()
    {
        removeAllJobs();
    }

    void removeAllJobs()
    {
        boost::mutex::scoped_lock l(mutex_);
        for (std::map<std::string, ScheduleOP>::iterator itr = jobs_.begin();
                itr != jobs_.end(); ++itr)
        {
            ScheduleOP *job = &itr->second;
            if (job->timer)
            {
                job->timer->stop();
                delete job->timer;
            }
        }
        jobs_.clear();
    }

    bool addJob(const string &name, uint32_t default_interval,
                uint32_t delay_start, const boost::function<void (int)>& func)
    {
        boost::mutex::scoped_lock l(mutex_);
        std::map<string, ScheduleOP>::iterator find_itr = jobs_.find(name);
        if (find_itr != jobs_.end())
            return false;

        ScheduleOP newjob;
        newjob.name = name;
        newjob.default_interval = default_interval;
        newjob.delay_start = delay_start;
        newjob.callback = func;
        newjob.timer = NULL;
        newjob.running = false;

        std::pair<std::map<std::string, ScheduleOP>::iterator, bool> result
        = jobs_.insert(make_pair(name, newjob));
        if (!result.second)
        {
            return false;
        }
        ScheduleOP *job = &result.first->second;

        uint32_t delay = job->delay_start;
        job->timer = new QueueTimer(&timerCallback, job, delay,
                                    job->default_interval);
        if (job->timer == NULL)
        {
            return false;
        }
        const bool started =  job->timer->start();
        if (started)
        {
            return true;
        }
        else
        {
            delete job->timer;
            return false;
        }
    }

    bool runJobImmediatly(const std::string& name, int calltype, bool sync)
    {
        boost::mutex::scoped_lock l(mutex_);
        std::map<std::string, ScheduleOP>::iterator itr = jobs_.find(name);
        if (itr == jobs_.end())
        {
            std::cout << "schedule job not found:" << name << std::endl;
            return false;
        }
        ScheduleOP *job = &itr->second;
        if (job && job->timer)
        {
            int retry = 5;
            while (job->running)
            {
                if (retry-- < 0)
                {
                    std::cout << "schedule job already running:" << name << std::endl;
                    return false;
                }
                sleep(1);
            }
            job->running = true;
            job->callback(calltype);
            job->running = false;
            return true;
        }
        std::cout << "schedule job timer null:" << name << std::endl;
        return false;
    }

    bool removeJob(const std::string &name)
    {
        boost::mutex::scoped_lock l(mutex_);
        std::map<std::string, ScheduleOP>::iterator itr = jobs_.find(name);
        if (itr == jobs_.end())
        {
            return false;
        }
        else
        {
            ScheduleOP *job = &itr->second;
            if (job->timer != NULL)
            {
                job->timer->stop();
                delete job->timer;
            }
            jobs_.erase(itr);
            return true;
        }
    }

private:
    static void timerCallback(void *param)
    {
        ScheduleOP *job = reinterpret_cast<ScheduleOP *>(param);
        if (job->running)
            return;
        job->running = true;

        job->callback(0);
        job->running = false;
    }

    std::map<std::string, ScheduleOP> jobs_;
    boost::mutex mutex_;
};

bool Scheduler::addJob(const string &name, uint32_t default_interval,
                       uint32_t delay_start, const boost::function<void (int)>& func)
{
    return Singleton<SchedulerImpl>::get()->addJob(name, default_interval, delay_start, func);
}

bool Scheduler::removeJob(const string &name)
{
    return Singleton<SchedulerImpl>::get()->removeJob(name);
}

void Scheduler::removeAllJobs()
{'fraction_non_alphanumeric': 0.12602659366445051, 'fraction_numerical': 0.005181853734845522, 'mean_word_length': 1.8358746881064596, 'pattern_counts': {'":': 0, '<': 78, '<?xml version=': 0, '>': 107, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4593967', 'n_tokens_mistral': 3081, 'n_tokens_neox': 2990, 'n_words': 654}