Add multiple features for jobs engine like parent child dependencies, throttling with sleep between requests, timeout for processing

Author: herrcristi

examples/examples_jobs_engine.h

@@ -127,7 +127,6 @@ namespace examples::jobs_engine {
 
             // set a custom delay (timeout for job3 is 500 ms)
             jobs_config.m_delay_next_request = std::chrono::milliseconds(500);
-            small::sleep(500); // TODO remove this after delay works
         });
 
         // add specific function for job1 (calling the function from jobs intead of config allows to pass the engine and extra param)
@@ -184,8 +183,7 @@ namespace examples::jobs_engine {
 
                 first_job = false;
             }
-            // TODO config to wait after request (even if it is not specified in the global config - so custom throttle)
-            small::sleep(30);
+            // config to wait after request (even if it is not specified in the global config - so custom throttle)
             jobs_config.m_delay_next_request = std::chrono::milliseconds(30);
         });
 

include/impl/jobs_engine_thread_pool_impl.h

@@ -73,7 +73,7 @@ namespace small::jobsimpl {
             // even if here it is considered that there are items and something will be scheduled,
             // the actual check if work will still exists will be done in do_action of parent
             auto &stats = it->second;
-            jobs_action_start(job_group, true, stats);
+            jobs_action_start(job_group, true /*has items*/, std::chrono::milliseconds(0) /*delay*/, stats);
         }
 
         // clang-format off
@@ -129,7 +129,7 @@ namespace small::jobsimpl {
         //
         // to trigger action (if needed for the new job group)
         //
-        inline void jobs_action_start(const JobGroupT &job_group, const bool has_items, JobGroupStats &stats)
+        inline void jobs_action_start(const JobGroupT &job_group, const bool has_items, const std::chrono::milliseconds &delay_next_request, JobGroupStats &stats)
         {
             if (!has_items) {
                 return;
@@ -141,14 +141,18 @@ namespace small::jobsimpl {
             bool needs_runners = stats.m_running < stats.m_threads_count;
             if (needs_runners) {
                 ++stats.m_running;
-                m_workers.push_back(job_group);
+                if (delay_next_request.count() > 0) {
+                    m_workers.push_back_delay_for(delay_next_request, job_group);
+                } else {
+                    m_workers.push_back(job_group);
+                }
             }
         }
 
         //
         // job action ended
         //
-        inline void jobs_action_end(const JobGroupT &job_group, const bool has_items)
+        inline void jobs_action_end(const JobGroupT &job_group, const bool has_items, const std::chrono::milliseconds &delay_next_request)
         {
             auto it = m_scheduler.find(job_group); // map is not changed, so can be access without locking
             if (it == m_scheduler.end()) {
@@ -160,7 +164,7 @@ namespace small::jobsimpl {
             auto &stats = it->second;
             --stats.m_running;
 
-            jobs_action_start(job_group, has_items, stats);
+            jobs_action_start(job_group, has_items, delay_next_request, stats);
         }
 
         //
@@ -170,11 +174,21 @@ namespace small::jobsimpl {
         {
             for (auto job_group : items) {
 
-                bool has_items = false;
-                m_parent_caller.do_action(job_group, &has_items);
+                std::chrono::milliseconds delay_next_request{};
+
+                auto ret       = m_parent_caller.do_action(job_group, delay_next_request);
+                bool has_items = ret == small::EnumLock::kElement;
+
+                // has items does not mean that there are still items in the queue
+                // it means that items were processed in this iteration
+                // as such an execution will be scheduled no matter what
+                // and next processing time will check to see if there are items or not
+
+                // for delay is important items -> delay -> next item -> delay ....
+                //                                       -> empty ... when new items arrive delay was already processed
 
                 // start another action
-                jobs_action_end(job_group, has_items);
+                jobs_action_end(job_group, has_items, delay_next_request);
             }
         }
 

include/impl/jobs_queue_impl.h

@@ -517,6 +517,17 @@ namespace small::jobsimpl {
         //
         // get job items
         //
+        inline std::shared_ptr<JobsItem> *jobs_get(const JobsID &jobs_id)
+        {
+            std::unique_lock l(m_lock);
+
+            auto it_j = m_jobs.find(jobs_id);
+            if (it_j == m_jobs.end()) {
+                return nullptr;
+            }
+            return &it_j->second;
+        }
+
         inline std::vector<std::shared_ptr<JobsItem>> jobs_get(const std::vector<JobsID> &jobs_ids)
         {
             std::vector<std::shared_ptr<JobsItem>> jobs_items;
@@ -525,27 +536,15 @@ namespace small::jobsimpl {
             std::unique_lock l(m_lock);
 
             for (auto &jobs_id : jobs_ids) {
-                auto it_j = m_jobs.find(jobs_id);
-                if (it_j == m_jobs.end()) {
-                    continue;
+                auto *jobs_item = jobs_get(jobs_id);
+                if (jobs_item) {
+                    jobs_items.push_back(*jobs_item);
                 }
-                jobs_items.push_back(it_j->second);
             }
 
             return jobs_items; // will be moved
         }
 
-        inline std::shared_ptr<JobsItem> *jobs_get(const JobsID &jobs_id)
-        {
-            std::unique_lock l(m_lock);
-
-            auto it_j = m_jobs.find(jobs_id);
-            if (it_j == m_jobs.end()) {
-                return nullptr;
-            }
-            return &it_j->second;
-        }
-
         //
         // add jobs item
         //

include/jobs_engine.h

@@ -431,15 +431,11 @@ namespace small {
         }
 
         //
-        // inner thread function for executing items (should return if there are more items) (called from thread_pool)
+        // get jobs to execute based on the group
         //
-        friend small::jobsimpl::jobs_engine_thread_pool<JobsGroupT, ThisJobsEngine>;
-
-        inline EnumLock do_action(const JobsGroupT &jobs_group, bool *has_items)
+        inline EnumLock get_group_jobs(const JobsGroupT &jobs_group, std::vector<JobsID> &vec_ids, typename JobsConfig::ConfigProcessing &group_config)
         {
-            *has_items = false;
-
-            // get bulk_count
+            // get bulk_count property
             auto it_cfg_grp = m_config.m_groups.find(jobs_group);
             if (it_cfg_grp == m_config.m_groups.end()) {
                 return small::EnumLock::kExit;
@@ -448,7 +444,6 @@ namespace small {
             int bulk_count = std::max(it_cfg_grp->second.m_bulk_count, 1);
 
             // delay request
-            typename JobsConfig::ConfigProcessing group_config{};
             group_config.m_delay_next_request = it_cfg_grp->second.m_delay_next_request;
 
             // get items to process
@@ -457,40 +452,53 @@ namespace small {
                 return small::EnumLock::kExit;
             }
 
-            std::vector<JobsID> vec_ids;
-            auto                ret = q->wait_pop_front_for(std::chrono::nanoseconds(0), vec_ids, bulk_count);
+            auto ret = q->wait_pop_front_for(std::chrono::nanoseconds(0), vec_ids, bulk_count);
+            return ret;
+        }
+
+        //
+        // inner thread function for executing items (should return if there are more items) (called from thread_pool)
+        //
+        friend small::jobsimpl::jobs_engine_thread_pool<JobsGroupT, ThisJobsEngine>;
+
+        inline EnumLock do_action(const JobsGroupT &jobs_group, std::chrono::milliseconds &delay_next_request)
+        {
+            // get jobs for the group
+            typename JobsConfig::ConfigProcessing group_config{}; // for delay request
+            std::vector<JobsID>                   vec_ids;
+
+            auto ret = get_group_jobs(jobs_group, vec_ids, group_config);
             if (ret != small::EnumLock::kElement) {
                 return ret;
             }
 
-            *has_items = true;
-
             // split by type
-            std::unordered_map<JobsTypeT, std::vector<std::shared_ptr<JobsItem>>> elems_by_type;
+            std::unordered_map<JobsTypeT, std::vector<std::shared_ptr<JobsItem>>> jobs_in_progress_by_type;
             {
                 // get jobs
-                std::vector<std::shared_ptr<JobsItem>> jobs_items = m_queue.jobs_get(vec_ids);
+                std::vector<std::shared_ptr<JobsItem>> jobs_items = queue().jobs_get(vec_ids);
                 for (auto &jobs_item : jobs_items) {
-                    elems_by_type[jobs_item->m_type].reserve(jobs_items.size());
+                    jobs_in_progress_by_type[jobs_item->m_type].reserve(jobs_items.size());
 
                     // mark the item as in progress
                     jobs_item->set_state_inprogress();
-                    // execute if it is still in progress (may be moved to higher states due to external factors like cancel, timeout, finish early due to other job, etc)
+                    // execute if it is still in progress
+                    // (may be moved to higher states due to external factors like cancel, timeout, finish early due to other job, etc)
                     if (jobs_item->is_state_inprogress()) {
-                        elems_by_type[jobs_item->m_type].push_back(jobs_item);
+                        jobs_in_progress_by_type[jobs_item->m_type].push_back(jobs_item);
                     }
                 }
             }
 
             // process specific job by type
-            for (auto &[jobs_type, jobs] : elems_by_type) {
+            for (auto &[jobs_type, jobs] : jobs_in_progress_by_type) {
                 auto it_cfg_type = m_config.m_types.find(jobs_type);
                 if (it_cfg_type == m_config.m_types.end()) {
                     continue;
                 }
 
                 // process specific jobs by type
-                typename JobsConfig::ConfigProcessing type_config;
+                typename JobsConfig::ConfigProcessing type_config{};
                 it_cfg_type->second.m_function_processing(jobs, type_config);
 
                 // get the max for config
@@ -503,13 +511,13 @@ namespace small {
                 }
 
                 // mark the item as in wait for children of finished
-                // if in callback the state is set to failed, cancelled or timeout setting to finish wont succeed because if less value than those
+                // if in callback the state is set to failed, cancelled or timeout
+                // setting to finish wont succeed because is less value than those
                 jobs_waitforchildren(jobs);
             }
 
-            // TODO  group_config.m_delay_next_request
-            // TODO for delay after requests use worker_thread delay item -> check if has_items should be set properly
-            // TODO if delay is set set has_items to true to force the sleep, but also a last time sleep so if there too much time and are no items dont continue
+            // delay request
+            delay_next_request = group_config.m_delay_next_request.value_or(std::chrono::milliseconds(0));
 
             return ret;
         }