Skip to content

Commit

Permalink
Add multiple features for jobs engine like parent child dependencies,…
Browse files Browse the repository at this point in the history 
… throttling with sleep between requests, timeout for processing
  • Loading branch information
@herrcristi
herrcristi committed Jan 14, 2025
1 parent d3993dc commit 253b16e
Show file tree
Hide file tree
Showing 5 changed files with 137 additions and 112 deletions.
1 change: 1 addition & 0 deletions .clang-format
View file
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@ AlignConsecutiveAssignments:
AlignFunctionDeclarations: true
AlignConsecutiveBitFields: true
AlignConsecutiveMacros: true
# PointerAlignment: Left
BraceWrapping:
AfterEnum: true
AfterStruct: true
Expand Down
100 changes: 53 additions & 47 deletions examples/examples_jobs_engine.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -77,7 +77,7 @@ namespace examples::jobs_engine {

.m_types = {{JobsType::kJobsType1, {.m_group = JobsGroupType::kJobsGroup12}},
{JobsType::kJobsType2, {.m_group = JobsGroupType::kJobsGroup12}},
{JobsType::kJobsType3, {.m_group = JobsGroupType::kJobsGroup3, .m_timeout = std::chrono::milliseconds(500)}},
{JobsType::kJobsType3, {.m_group = JobsGroupType::kJobsGroup3, .m_timeout = std::chrono::milliseconds(700)}},
{JobsType::kJobsDatabase, {.m_group = JobsGroupType::kJobsGroupDatabase}},
{JobsType::kJobsCache, {.m_group = JobsGroupType::kJobsGroupCache}}}};

Expand All @@ -88,53 +88,52 @@ namespace examples::jobs_engine {

// create a cache server (with workers to simulate access to it)
// (as an external engine outside the jobs engine for demo purposes)
small::worker_thread<JobsEng::JobsID> cache_server({.threads_count = 1}, [&](auto &w /*this*/, const auto &items) {
for (auto &i : items) {
std::cout << "thread " << std::this_thread::get_id()
<< " CACHE processing"
<< " {" << i << "}" << "\n";
small::worker_thread<std::pair<JobsEng::JobsID, Request>> cache_server({.threads_count = 1}, [&](auto &w /*this*/, const auto &items) {
for (auto &[job_id, req] : items) {
std::cout << "worker thread " << std::this_thread::get_id()
<< " CACHE processing"
<< " {" << req.first << ", " << req.second << ", jobid=" << job_id << "}"
<< " time " << small::toISOString(small::timeNow())
<< "\n";

// mark the jobs id associated as succeeded (for demo purposes to avoid creating other structures)
jobs.jobs_finished(i, (Response)i);
jobs.jobs_finished(job_id, (Response)job_id);
}
// sleep long enough
// no coalesce for demo purposes (sleep 500) so 3rd parent items is finished due to database and not cache server
small::sleep(500);
});

auto fn_print_item = [](auto item, std::string fn_type) {
std::cout << "thread " << std::this_thread::get_id()
<< std::setw(10) << fn_type
<< " processing "
<< "{"
<< " jobid=" << std::setw(2) << item->m_id
<< " type=" << std::setw(1) << (int)item->m_type
<< " req.int=" << std::setw(2) << item->m_request.first << ","
<< " req.str=\"" << item->m_request.second << "\""
<< "}"
<< " time " << small::toISOString(small::timeNow())
<< "\n";
};

// default processing used for job type 3 with custom delay in between requests
// one request will succeed and one request will timeout for demo purposes
jobs.config_default_function_processing([](auto &j /*this jobs engine*/, const auto &jobs_items, auto &jobs_config) {
jobs.config_default_function_processing([&](auto &j /*this jobs engine*/, const auto &jobs_items, auto &jobs_config) {
for (auto &item : jobs_items) {
std::cout << "thread " << std::this_thread::get_id()
<< " DEFAULT processing "
<< "{"
<< " type=" << (int)item->m_type
<< " req.int=" << item->m_request.first << ","
<< " req.str=\"" << item->m_request.second << "\""
<< "}"
<< " ref count " << item.use_count()
<< " time " << small::toISOString(small::timeNow())
<< "\n";
fn_print_item(item, "DEFAULT");
}

// set a custom delay (timeout for job3 is 500 ms)
jobs_config.m_delay_next_request = std::chrono::milliseconds(1000);
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)
jobs.config_jobs_function_processing(JobsType::kJobsType1, [&](auto &j /*this jobs engine*/, const auto &jobs_items, auto & /* config */, auto b /*extra param b*/) {
for (auto &item : jobs_items) {
std::cout << "thread " << std::this_thread::get_id()
<< " JOB1 processing "
<< "{"
<< " type=" << (int)item->m_type
<< " req.int=" << item->m_request.first << ","
<< " req.str=\"" << item->m_request.second << "\""
<< "}"
<< " ref count " << item.use_count()
<< " time " << small::toISOString(small::timeNow())
<< "\n";
fn_print_item(item, "JOB1");

// add 2 more children jobs for current one for database and server cache
JobsEng::JobsID jobs_child_db_id{};
Expand All @@ -152,7 +151,7 @@ namespace examples::jobs_engine {

j.jobs_start(small::EnumPriorities::kNormal, jobs_child_db_id);
// jobs_child_cache_id has no threads to execute, it has external executors
cache_server.push_back(jobs_child_cache_id);
cache_server.push_back({jobs_child_cache_id, item->m_request});
}
small::sleep(30); }, 5 /*param b*/);

Expand All @@ -165,25 +164,16 @@ namespace examples::jobs_engine {
// TODO set state merge daca e doar o dependinta, daca sunt mai multe atunci ar tb o functie custom - childProcessing (desi are sau nu are children - sau cum fac un dummy children - poate cu thread_count 0?)

// add specific function for job2
jobs.config_jobs_function_processing(JobsType::kJobsType2, [](auto &j /*this jobs engine*/, const auto &jobs_items, auto & /* config */) {
bool first_job = true;
jobs.config_jobs_function_processing(JobsType::kJobsType2, [&](auto &j /*this jobs engine*/, const auto &jobs_items, auto &jobs_config) {
static bool first_job = true;
for (auto &item : jobs_items) {
std::cout << "thread " << std::this_thread::get_id()
<< " JOB2 processing "
<< "{"
<< " type=" << (int)item->m_type
<< " req.int=" << item->m_request.first << ","
<< " req.str=\"" << item->m_request.second << "\""
<< "}"
<< " ref count " << item.use_count()
<< " time " << small::toISOString(small::timeNow())
<< "\n";
fn_print_item(item, "JOB2");

if (first_job) {
// for type 2 only database children (for demo purposes no result will be used from database)
auto ret = j.queue().push_back_and_start_child(item->m_id /*parent*/,
small::EnumPriorities::kNormal,
JobsType::kJobsDatabase,
small::EnumPriorities::kNormal,
JobsType::kJobsDatabase,
item->m_request);
if (!ret) {
j.jobs_failed(item->m_id);
Expand All @@ -195,7 +185,21 @@ 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); });
small::sleep(30);
jobs_config.m_delay_next_request = std::chrono::milliseconds(30);
});

// add specific function for job2
jobs.config_jobs_function_processing(JobsType::kJobsDatabase, [&](auto &j /*this jobs engine*/, const auto &jobs_items, auto & /* config */) {
for (auto &item : jobs_items) {
// simulate long db call
small::sleep(200);

fn_print_item(item, "DATABASE");

// this job will be auto-finished
}
});

// TODO add function for database where demonstrate coalesce of 3 items (sleep 1000)
// TODO add function for cache server - no coalesce for demo purposes (sleep 500) so 3rd parent items is finished due to database and not cache server
Expand All @@ -207,8 +211,8 @@ namespace examples::jobs_engine {
std::vector<JobsEng::JobsID> jobs_ids;

// type3 one request will succeed and one request will timeout for demo purposes
jobs.queue().push_back_and_start(small::EnumPriorities::kNormal, JobsType::kJobsType3, {3, "normal3"}, &jobs_id);
jobs.queue().push_back_and_start(small::EnumPriorities::kHigh, JobsType::kJobsType3, {3, "high3"}, &jobs_id);
jobs.queue().push_back_and_start_delay_for(std::chrono::milliseconds(100), small::EnumPriorities::kNormal, JobsType::kJobsType3, {3, "normal3"}, &jobs_id);
jobs.queue().push_back_and_start_delay_for(std::chrono::milliseconds(100), small::EnumPriorities::kHigh, JobsType::kJobsType3, {3, "high3"}, &jobs_id);

// type2 only the first request succeeds and waits for child the other fails from the start
jobs.queue().push_back_and_start(small::EnumPriorities::kNormal, JobsType::kJobsType2, {2, "normal2"}, &jobs_id);
Expand Down Expand Up @@ -242,6 +246,8 @@ namespace examples::jobs_engine {
auto ret = jobs.wait_for(std::chrono::milliseconds(100)); // wait to finished
std::cout << "wait for with timeout, ret = " << static_cast<int>(ret) << " as timeout\n";
jobs.wait(); // wait here for jobs to finish due to exit flag
cache_server.signal_exit_force();
cache_server.wait();

std::cout << "size = " << jobs.size() << "\n";

Expand Down
1 change: 1 addition & 0 deletions include/impl/jobs_item_impl.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -101,6 +101,7 @@ namespace small::jobsimpl {
inline void set_state_cancelled () { set_state(EnumJobsState::kCancelled); }

inline bool is_state (const EnumJobsState &state) { return m_state.load() == state; }
static bool is_state_complete (const EnumJobsState &state) { return state >= EnumJobsState::kFinished; }

inline bool is_state_inprogress () { return is_state(EnumJobsState::kInProgress); }
inline void is_state_waitchildren () { return is_state(EnumJobsState::kWaitChildren); }
Expand Down
53 changes: 10 additions & 43 deletions include/impl/jobs_queue_impl.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@

namespace small::jobsimpl {
//
// small queue helper class for jobs (parent caller must implement 'jobs_schedule', 'jobs_finished')
// small queue helper class for jobs (parent caller must implement 'jobs_add', 'jobs_schedule', 'jobs_finished')
//
template <typename JobsTypeT, typename JobsRequestT, typename JobsResponseT, typename JobsGroupT, typename JobsPrioT, typename ParentCallerT>
class jobs_queue
Expand Down Expand Up @@ -94,17 +94,14 @@ namespace small::jobsimpl {
// config job types
// m_types_queues will be initialized in the initial setup phase and will be accessed without locking afterwards
//
inline bool config_jobs_type(const JobsTypeT &jobs_type, const JobsGroupT &jobs_group, const std::optional<std::chrono::milliseconds> &jobs_timeout)
inline bool config_jobs_type(const JobsTypeT &jobs_type, const JobsGroupT &jobs_group)
{
auto it_g = m_groups_queues.find(jobs_group);
if (it_g == m_groups_queues.end()) {
return false;
}

m_types_queues[jobs_type] = &it_g->second;
if (jobs_timeout) {
m_types_timeouts[jobs_type] = *jobs_timeout;
}
return true;
}

Expand Down Expand Up @@ -562,8 +559,9 @@ namespace small::jobsimpl {
jobs_item->m_id = id;
m_jobs.emplace(id, jobs_item);

// add it to the timeout queue
jobs_start_timeout(jobs_item);
// call parent for extra processing
m_parent_caller.jobs_add(jobs_item);

return id;
}

Expand Down Expand Up @@ -610,21 +608,6 @@ namespace small::jobsimpl {
return ret;
}

//
// add it to the timeout queue
//
inline std::size_t jobs_start_timeout(std::shared_ptr<JobsItem> jobs_item)
{
std::unique_lock l(m_lock);

// only if job type has config a timeout
auto it_timeout = m_types_timeouts.find(jobs_item->m_type);
if (it_timeout == m_types_timeouts.end()) {
return 0;
}
return m_timeout_queue.queue().push_delay_for(it_timeout->second, jobs_item->m_id);
}

//
// erase jobs item
//
Expand Down Expand Up @@ -685,34 +668,18 @@ namespace small::jobsimpl {
return count;
}

//
// inner thread function for timeout items
// called from m_timeout_queue
//
using JobsQueueTimeout = small::time_queue_thread<JobsID, ThisJobsQueue>;
friend JobsQueueTimeout;

inline std::size_t push_back(std::vector<JobsID> &&jobs_ids)
{
m_parent_caller.jobs_timeout(jobs_ids);
return jobs_ids.size();
}

private:
//
// members
//
mutable small::base_lock m_lock; // global locker
std::atomic<JobsID> m_jobs_seq_id{}; // to get the next jobs id
std::unordered_map<JobsID, std::shared_ptr<JobsItem>> m_jobs; // current jobs
std::unordered_map<JobsGroupT, JobsQueue> m_groups_queues; // map of queues by group
std::unordered_map<JobsTypeT, JobsQueue *> m_types_queues; // optimize to have queues by type (which reference queues by group)
std::unordered_map<JobsTypeT, std::chrono::milliseconds> m_types_timeouts; // timeouts for types
mutable small::base_lock m_lock; // global locker
std::atomic<JobsID> m_jobs_seq_id{}; // to get the next jobs id
std::unordered_map<JobsID, std::shared_ptr<JobsItem>> m_jobs; // current jobs
std::unordered_map<JobsGroupT, JobsQueue> m_groups_queues; // map of queues by group
std::unordered_map<JobsTypeT, JobsQueue *> m_types_queues; // optimize to have queues by type (which reference queues by group)

JobQueueDelayedT m_delayed_items{*this}; // queue of delayed items

JobsQueueTimeout m_timeout_queue{*this}; // for timeout elements

ParentCallerT &m_parent_caller; // jobs engine
};
} // namespace small::jobsimpl
Loading

0 comments on commit 253b16e

Please sign in to comment.