fixed bug in getjob where it could return null pointer

Author: brichards64

src/DataModelBase/WorkerPoolManager.cpp

@@ -18,7 +18,7 @@ PoolManager_args::PoolManager_args() : Thread_args() {}
 
 PoolManager_args::~PoolManager_args() {}
 
-WorkerPoolManager::WorkerPoolManager(JobQueue& job_queue, unsigned int* thread_cap, unsigned int* global_thread_cap, unsigned int* global_thread_num, JobDeque* job_out_deque, bool self_serving, bool threaded, unsigned int thread_sleep_us, unsigned int thread_management_period_us, unsigned int job_assignment_period_us){
+WorkerPoolManager::WorkerPoolManager(JobQueue& job_queue, unsigned int* thread_cap, unsigned int* global_thread_cap, std::atomic<unsigned int>* global_thread_num, JobDeque* job_out_deque, bool self_serving, bool threaded, unsigned int thread_sleep_us, unsigned int thread_management_period_us, unsigned int job_assignment_period_us){
 
   m_util = new Utilities();
 
@@ -69,7 +69,7 @@ void WorkerPoolManager::CreateManagerThread() {
 }
 
 
-void WorkerPoolManager::CreateWorkerThread(std::vector<PoolWorker_args*>& in_args, bool &in_self_serving, unsigned int &in_thread_sleep_us, JobQueue* in_job_queue, JobDeque* in_job_out_deque,unsigned long &thread_num, Utilities* in_util, std::map<std::string,PoolManagerStats>* in_stats, std::mutex* in_stats_mtx, unsigned int* global_thread_num) {
+void WorkerPoolManager::CreateWorkerThread(std::vector<PoolWorker_args*>& in_args, bool &in_self_serving, unsigned int &in_thread_sleep_us, JobQueue* in_job_queue, JobDeque* in_job_out_deque,unsigned long &thread_num, Utilities* in_util, std::map<std::string,PoolManagerStats>* in_stats, std::mutex* in_stats_mtx, std::atomic<unsigned int>* global_thread_num) {
   PoolWorker_args* tmparg = new PoolWorker_args();
   tmparg->busy = false;
   tmparg->thread_sleep_us = in_thread_sleep_us;
@@ -88,12 +88,12 @@ void WorkerPoolManager::CreateWorkerThread(std::vector<PoolWorker_args*>& in_arg
   if(global_thread_num) (*global_thread_num)++;
 }
 
-void WorkerPoolManager::DeleteWorkerThread(unsigned int pos,  Utilities* in_util, std::vector<PoolWorker_args*> &in_args, unsigned int* global_thread_num) {
+void WorkerPoolManager::DeleteWorkerThread(unsigned int pos,  Utilities* in_util, std::vector<PoolWorker_args*> &in_args, std::atomic<unsigned int>* global_thread_num) {
   in_util->KillThread(in_args.at(pos));
   delete in_args.at(pos);
   in_args.at(pos) = 0;
   in_args.erase(in_args.begin() + pos );
-  if(global_thread_num) global_thread_num--;
+  if(global_thread_num) (*global_thread_num)--;
 }
 
 void WorkerPoolManager::WorkerThread(Thread_args* arg) {
@@ -103,6 +103,11 @@ void WorkerPoolManager::WorkerThread(Thread_args* arg) {
   else {
     if(args->self_serving){
       args->job=args->job_queue->GetJob();
+      if(!args->job){
+	usleep(args->thread_sleep_us);
+	return;
+      }
+
       args->stats_mtx->lock();
       (*args->stats)[args->job->m_id].processing++;
       args->stats_mtx->unlock();
@@ -154,7 +159,7 @@ void WorkerPoolManager::WorkerThread(Thread_args* arg) {
     if (args->job_out_deque || args->job->out_deque) {
       if(args->job->out_deque) args->job->out_deque->push_back(args->job);
       else args->job_out_deque->push_back(args->job);
-     args->job->m_in_progress=false;
+      args->job->m_in_progress=false;
     } 
     else if(args->job->out_pool){
       args->job->out_pool->Add(args->job);
@@ -172,11 +177,11 @@ void WorkerPoolManager::WorkerThread(Thread_args* arg) {
 void WorkerPoolManager::ManagerThread(Thread_args* arg) {
 
   PoolManager_args* args = reinterpret_cast<PoolManager_args*>(arg);
-
+ 
   args->now = std::chrono::high_resolution_clock::now();
   args->manage = std::chrono::duration<double, std::micro>(args->now - args->managing_timer).count() > args->thread_management_period_us;
   args->sleep = !args->manage;
-
+  
   if(!args->self_serving){
     args->serve = std::chrono::duration<double, std::micro>(args->now - args->serving_timer).count() > args->job_assignment_period_us;
     args->sleep = !args->serve && !args->manage;  
@@ -188,11 +193,11 @@ void WorkerPoolManager::ManagerThread(Thread_args* arg) {
   }
 
   if(args->serve){
-   
     if (args->job_queue->size() > 0) {
       for (unsigned int i = 0; i < args->args.size(); i++) {
 	if (!args->args.at(i)->busy && args->job_queue->size() > 0) {
-	  args->args.at(i)->job = args->job_queue->GetJob();
+	  args->args.at(i)->job = args->job_queue->GetJob(); 
+	  if(args->args.at(i)->job == 0) continue;
 	  args->stats_mtx.lock();
 	  args->stats[args->args.at(i)->job->m_id].processing++;
 	  args->stats_mtx.unlock();
@@ -206,7 +211,6 @@ void WorkerPoolManager::ManagerThread(Thread_args* arg) {
   }
 
   if(args->manage){
-
     args->free_threads = 0;
     unsigned int last_free = 0;
     for (unsigned int i = 0; i < args->args.size(); i++) {
@@ -216,8 +220,8 @@ void WorkerPoolManager::ManagerThread(Thread_args* arg) {
       }
     }
 
-    if (args->free_threads < 1 && args->args.size()<(*(args->thread_cap)) && ( !args->global_thread_cap || (*(args->global_thread_num))<(*(args->global_thread_cap))  ) ) CreateWorkerThread(args->args, args->self_serving, args->thread_sleep_us, args->job_queue, args->job_out_deque, args->thread_num, args->util, &args->stats, &args->stats_mtx, args->global_thread_num);
-    
+    if (args->free_threads < 1 && args->args.size()<(*(args->thread_cap)) && ( !args->global_thread_cap || (*(args->global_thread_num))<(*(args->global_thread_cap))  ) )       CreateWorkerThread(args->args, args->self_serving, args->thread_sleep_us, args->job_queue, args->job_out_deque, args->thread_num, args->util, &args->stats, &args->stats_mtx, args->global_thread_num);
+
     if (args->free_threads > 1) DeleteWorkerThread(last_free, args->util, args->args, args->global_thread_num);
 
     args->managing_timer = std::chrono::high_resolution_clock::now();
@@ -243,7 +247,7 @@ std::string WorkerPoolManager::GetStats(){
 
   std::string ret="";
 
-  ret="Queued Jobs Total = " + std::to_string(m_job_queue->size()) + " \n"; 
+  ret="Queued Jobs Total = " + std::to_string(m_job_queue->size()) + " : Total Workers = " + std::to_string(NumThreads()) + " \n"; 
   m_job_queue->m_lock.lock();
   m_manager_args.stats_mtx.lock(); 
 

src/DataModelBase/WorkerPoolManager.h

@@ -9,6 +9,7 @@
 #include <Utilities.h>
 #include <mutex>
 #include <chrono>
+#include <atomic>
 
 namespace ToolFramework{
 
@@ -72,7 +73,7 @@ namespace ToolFramework{
     JobQueue* job_queue;
     unsigned int* thread_cap;  ///< Variable to cap the number of thread workers
     unsigned int* global_thread_cap;
-    unsigned int* global_thread_num;
+    std::atomic<unsigned int>* global_thread_num;
     JobDeque* job_out_deque;
     bool self_serving;
     unsigned int thread_sleep_us;
@@ -120,7 +121,7 @@ namespace ToolFramework{
        * @param thrad_management_period_us how long between evaluating the number of worker threads to avoid rapid killing and recreating
        * @param job_assignment_period how long the managers sleeps between checking if there are free workers to assign them new jobs
        */
-    WorkerPoolManager(JobQueue& job_queue,  unsigned int* thread_cap=0, unsigned int* global_thread_cap=0, unsigned int* global_thread_num=0, JobDeque* job_out_deque=0, bool self_serving=false, bool threaded=true, unsigned int thread_sleep_us=100, unsigned int thread_management_period_us=10000, unsigned int job_assignment_period_us=1000);
+    WorkerPoolManager(JobQueue& job_queue,  unsigned int* thread_cap=0, unsigned int* global_thread_cap=0, std::atomic<unsigned int>* global_thread_num=0, JobDeque* job_out_deque=0, bool self_serving=false, bool threaded=true, unsigned int thread_sleep_us=100, unsigned int thread_management_period_us=10000, unsigned int job_assignment_period_us=1000);
     ~WorkerPoolManager(); ///< Simple Destructor
 
     void ManageWorkers(); ///< Function to manage workers and distribute jobs to be run when unthreaded if you choose to not have the managment run on a thread.
@@ -132,8 +133,8 @@ namespace ToolFramework{
   private:
 
     void CreateManagerThread(); ///< Function to Create Manager Thread
-    static void CreateWorkerThread(std::vector<PoolWorker_args*> &in_args, bool &in_self_serving, unsigned int &in_thread_sleep_us, JobQueue* in_job_queue, JobDeque* in_job_out_deque,unsigned long &thread_num, Utilities* in_util, std::map<std::string,PoolManagerStats>* in_stats, std::mutex* in_stats_mtx, unsigned int* global_thread_num=0 ); ///< Function to Create Worker Thread
-    static void DeleteWorkerThread(unsigned int pos,  Utilities* in_util, std::vector<PoolWorker_args*> &in_args, unsigned int* global_thread_num=0); ///< Function to delete thread @param pos is the position in the args vector below
+    static void CreateWorkerThread(std::vector<PoolWorker_args*> &in_args, bool &in_self_serving, unsigned int &in_thread_sleep_us, JobQueue* in_job_queue, JobDeque* in_job_out_deque,unsigned long &thread_num, Utilities* in_util, std::map<std::string,PoolManagerStats>* in_stats, std::mutex* in_stats_mtx, std::atomic<unsigned int>* global_thread_num=0 ); ///< Function to Create Worker Thread
+    static void DeleteWorkerThread(unsigned int pos,  Utilities* in_util, std::vector<PoolWorker_args*> &in_args, std::atomic<unsigned int>* global_thread_num=0); ///< Function to delete thread @param pos is the position in the args vector below
 
     static void WorkerThread(Thread_args* arg); ///< Function to be run by the thread in a loop. Make sure not to block in it
     static void ManagerThread(Thread_args* arg); ///< Function to be run by the thread manager. Make sure not to block in it