Added a run_at, run_every and run_on_start, making the app a daemon like entity.
This commit is contained in:
parent
111c2487d8
commit
41c405a0c5
1 changed files with 694 additions and 46 deletions
|
|
@ -46,11 +46,25 @@ void ExecutionManager::setJobs(std::vector<std::shared_ptr<AppInfo> > &job_list)
|
|||
this->job_list = std::move(job_list);
|
||||
}
|
||||
|
||||
|
||||
//This function assumes that job_list_mutex is locked
|
||||
void ExecutionManager::printActiveJobs()
|
||||
{
|
||||
for(auto &i: job_list){
|
||||
log_file << "Main Thread: " + i->getModuleName() + " mode: " + i->getStringRunMode() + " is running.";
|
||||
}
|
||||
}
|
||||
|
||||
void ExecutionManager::run()
|
||||
{
|
||||
finished = false;
|
||||
initial_list_size = job_list.size();
|
||||
seperateScheduledTasks();
|
||||
log_file << "Main Thread: Job list: " << job_list.size() <<
|
||||
" Scheduled list: " << scheduled_list.size() << std::endl;
|
||||
last_poll_time = std::chrono::system_clock::now();
|
||||
while(true) {
|
||||
pollScheduledTasks();
|
||||
for(int i=active_thread_count.load(); i < threads ;i++) {
|
||||
int val = 0;
|
||||
{
|
||||
|
|
@ -61,7 +75,6 @@ void ExecutionManager::run()
|
|||
break;
|
||||
}
|
||||
if(val > (initial_list_size - 1)) {
|
||||
log_file << "Main Thread: All jobs queued." << std::endl;
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
|
@ -178,13 +191,21 @@ void ExecutionManager::run()
|
|||
default: {
|
||||
// After fork in parent do
|
||||
ptr->setRunning();
|
||||
// If it was a scheduled task recalculate its next execution time
|
||||
// and send it back to the scheduled job list
|
||||
if(ptr->getScheduled()) {
|
||||
ptr->addToPreviousExecutionList(pid);
|
||||
ptr->updateNextTimeToRun();
|
||||
std::lock_guard<std::mutex> lock_get(this->job_list_mutex);
|
||||
scheduled_list.push_back(ptr);
|
||||
}
|
||||
state_change.fetch_add(1);
|
||||
state_change.notify_all();
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
// If script daemon wait here
|
||||
// If script wait here
|
||||
AppInfo::RunMode mode = ptr->getRunMode();
|
||||
int status;
|
||||
if(mode == AppInfo::RunMode::script) {
|
||||
|
|
@ -198,7 +219,7 @@ void ExecutionManager::run()
|
|||
state_change.fetch_add(1);
|
||||
state_change.notify_all();
|
||||
os << std::vformat("Thread[{}] finished: ", std::make_format_args(val))
|
||||
<< module_name<< std::endl;
|
||||
<< module_name << std::endl;
|
||||
active_thread_count.fetch_sub(1);
|
||||
});
|
||||
thread_list.push_back(thread);
|
||||
|
|
@ -209,10 +230,6 @@ void ExecutionManager::run()
|
|||
log_file << "Main Thread: Job list is empty, waiting for finish." << std::endl;
|
||||
break;
|
||||
}
|
||||
if((index_count) > (initial_list_size - 1)) {
|
||||
log_file << "Main Thread: All jobs queued." << std::endl;
|
||||
break;
|
||||
}
|
||||
}
|
||||
if (active_thread_count.load() >= threads) {
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(10));
|
||||
|
|
@ -224,3 +241,48 @@ void ExecutionManager::run()
|
|||
}
|
||||
finished = true;
|
||||
}
|
||||
|
||||
bool ExecutionManager::checkScheduledTaskIsReady(std::chrono::system_clock::time_point next_time)
|
||||
{
|
||||
return (next_time <= std::chrono::system_clock::now());
|
||||
}
|
||||
|
||||
void ExecutionManager::seperateScheduledTasks()
|
||||
{
|
||||
for(auto task = job_list.begin(); task != job_list.end(); task++) {
|
||||
// If task is not running on start and task is scheduled.
|
||||
if(!((*task)->getRunOnStart()) && ((*task)->getScheduled())) {
|
||||
log_file << "Main Thread: Job[" + (*task)->getModuleName() +
|
||||
"] is scheduled for execution." << std::endl;
|
||||
scheduled_list.push_back(std::move(*task));
|
||||
task = job_list.erase(task);
|
||||
if(task == job_list.end())
|
||||
break;
|
||||
task--;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
void ExecutionManager::pollScheduledTasks()
|
||||
{
|
||||
auto current_poll_time = std::chrono::system_clock::now();
|
||||
if(std::chrono::duration_cast<std::chrono::seconds>(current_poll_time - last_poll_time).count() <= 1)
|
||||
return;
|
||||
last_poll_time = current_poll_time;
|
||||
{
|
||||
std::lock_guard<std::mutex> lock_get(this->job_list_mutex);
|
||||
log_file << "Main Thread: Polling before: Job list: " << job_list.size() <<
|
||||
" Scheduled list: " << scheduled_list.size() << std::endl;
|
||||
for(auto task = scheduled_list.begin(); task != scheduled_list.end(); task++) {
|
||||
if(checkScheduledTaskIsReady((*task)->getNextTimeToRun())) {
|
||||
job_list.push_back(std::move(*task));
|
||||
task = scheduled_list.erase(task);
|
||||
if(task == scheduled_list.end())
|
||||
break;
|
||||
task--;
|
||||
}
|
||||
}
|
||||
log_file << "Main Thread: Polling after: Job list: " << job_list.size() <<
|
||||
" Scheduled list: " << scheduled_list.size() << std::endl;
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue