Fixes for flaws found by gemini 3.8 on 21/10/26.
This commit is contained in:
parent
41c405a0c5
commit
bc5760eee9
1 changed files with 105 additions and 84 deletions
|
|
@ -58,45 +58,56 @@ void ExecutionManager::printActiveJobs()
|
|||
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) {
|
||||
// Clean up completed worker threads
|
||||
for(auto it = thread_list.begin(); it != thread_list.end(); ) {
|
||||
if((*it)->done.load()) {
|
||||
if((*it)->thread.joinable())
|
||||
(*it)->thread.join();
|
||||
it = thread_list.erase(it);
|
||||
} else {
|
||||
++it;
|
||||
}
|
||||
}
|
||||
|
||||
pollScheduledTasks();
|
||||
for(int i=active_thread_count.load(); i < threads ;i++) {
|
||||
|
||||
// Dispatch jobs from job_list to worker threads
|
||||
while(active_thread_count.load() < threads) {
|
||||
std::shared_ptr<AppInfo> ptr;
|
||||
int val = 0;
|
||||
size_t remaining_size = 0;
|
||||
{
|
||||
std::lock_guard<std::mutex> lock_get(this->job_list_mutex);
|
||||
val = index_count++;
|
||||
if(job_list.empty()) {
|
||||
log_file << "Main Thread: Job list is empty, waiting for finish." << std::endl;
|
||||
break;
|
||||
}
|
||||
if(val > (initial_list_size - 1)) {
|
||||
break;
|
||||
}
|
||||
ptr = job_list.front();
|
||||
job_list.erase(job_list.begin());
|
||||
remaining_size = job_list.size();
|
||||
val = index_count++;
|
||||
}
|
||||
|
||||
active_thread_count.fetch_add(1);
|
||||
std::shared_ptr<std::thread> thread = std::make_shared<std::thread>([this, val](){
|
||||
std::shared_ptr<AppInfo> ptr;
|
||||
std::osyncstream os(log_file.rdbuf());
|
||||
size_t list_size = 0;
|
||||
{
|
||||
std::lock_guard<std::mutex> lock_get(this->job_list_mutex);
|
||||
if(job_list.size() == 0) {
|
||||
os << std::vformat("Thread[{}] called while the job list was empty: ", std::make_format_args(val))
|
||||
<< std::endl;
|
||||
return;
|
||||
auto worker = std::make_shared<WorkerThread>();
|
||||
worker->thread = std::thread([this, val, ptr, remaining_size, worker](){
|
||||
struct ThreadGuard {
|
||||
std::atomic<int>& count;
|
||||
std::atomic<bool>& done;
|
||||
~ThreadGuard() {
|
||||
count.fetch_sub(1);
|
||||
done.store(true);
|
||||
}
|
||||
ptr = this->job_list[0];
|
||||
job_list.erase(job_list.begin());
|
||||
list_size = job_list.size();
|
||||
}
|
||||
} guard{active_thread_count, worker->done};
|
||||
|
||||
std::osyncstream os(log_file.rdbuf());
|
||||
std::string module_name = ptr->getModuleName();
|
||||
os << std::vformat("Thread[{}] running: ", std::make_format_args(val))
|
||||
<< module_name << " current list size: " << list_size << std::endl;
|
||||
<< module_name << " current list size: " << remaining_size << std::endl;
|
||||
int old = state_change.load();
|
||||
while(!ptr->isParentDependencyCleared(os))
|
||||
{
|
||||
|
|
@ -220,25 +231,26 @@ void ExecutionManager::run()
|
|||
state_change.notify_all();
|
||||
os << std::vformat("Thread[{}] finished: ", std::make_format_args(val))
|
||||
<< module_name << std::endl;
|
||||
active_thread_count.fetch_sub(1);
|
||||
});
|
||||
thread_list.push_back(thread);
|
||||
thread_list.push_back(worker);
|
||||
}
|
||||
|
||||
// Check if all work is done (no active jobs, no pending jobs, no scheduled tasks)
|
||||
{
|
||||
std::lock_guard<std::mutex> lock_get(this->job_list_mutex);
|
||||
if(job_list.empty()) {
|
||||
log_file << "Main Thread: Job list is empty, waiting for finish." << std::endl;
|
||||
if(job_list.empty() && scheduled_list.empty() && active_thread_count.load() == 0) {
|
||||
log_file << "Main Thread: All jobs finished." << std::endl;
|
||||
break;
|
||||
}
|
||||
}
|
||||
if (active_thread_count.load() >= threads) {
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(10));
|
||||
}
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(50));
|
||||
}
|
||||
for(auto i: thread_list) {
|
||||
if(i->joinable())
|
||||
i->join();
|
||||
|
||||
for(auto &worker: thread_list) {
|
||||
if(worker->thread.joinable())
|
||||
worker->thread.join();
|
||||
}
|
||||
thread_list.clear();
|
||||
finished = true;
|
||||
}
|
||||
|
||||
|
|
@ -249,16 +261,16 @@ bool ExecutionManager::checkScheduledTaskIsReady(std::chrono::system_clock::time
|
|||
|
||||
void ExecutionManager::seperateScheduledTasks()
|
||||
{
|
||||
for(auto task = job_list.begin(); task != job_list.end(); task++) {
|
||||
std::lock_guard<std::mutex> lock_get(this->job_list_mutex);
|
||||
for(auto task = job_list.begin(); task != job_list.end(); ) {
|
||||
// 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));
|
||||
scheduled_list.push_back(*task);
|
||||
task = job_list.erase(task);
|
||||
if(task == job_list.end())
|
||||
break;
|
||||
task--;
|
||||
} else {
|
||||
++task;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -266,23 +278,20 @@ void ExecutionManager::seperateScheduledTasks()
|
|||
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)
|
||||
if(std::chrono::duration_cast<std::chrono::seconds>(current_poll_time - last_poll_time).count() < polling_interval)
|
||||
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++) {
|
||||
for(auto task = scheduled_list.begin(); task != scheduled_list.end(); ) {
|
||||
if(checkScheduledTaskIsReady((*task)->getNextTimeToRun())) {
|
||||
job_list.push_back(std::move(*task));
|
||||
log_file << "Main Thread: Job[" + (*task)->getModuleName() +
|
||||
"] scheduled time arrived, adding to execution queue." << std::endl;
|
||||
job_list.push_back(*task);
|
||||
task = scheduled_list.erase(task);
|
||||
if(task == scheduled_list.end())
|
||||
break;
|
||||
task--;
|
||||
} else {
|
||||
++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