diff --git a/CMakeLists.txt b/CMakeLists.txt index 1baee4e..52ff948 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -1,16 +1,35 @@ -cmake_minimum_required(VERSION 3.16) +cmake_minimum_required(VERSION 3.30) project(StartUpRunner VERSION 1.0.0 LANGUAGES CXX) set(CMAKE_CXX_STANDARD 26) set(CMAKE_CXX_STANDARD_REQUIRED ON) +include(FetchContent) + +FetchContent_Declare( + absl + GIT_REPOSITORY https://github.com/abseil/abseil-cpp.git + GIT_TAG 20260526.0 # Use a stable version tag or commit hash +) +FetchContent_MakeAvailable(absl) +set(RE2_BUILD_TESTING OFF CACHE BOOL "" FORCE) +set(RE2_INSTALL OFF CACHE BOOL "" FORCE) +FetchContent_Declare( + re2 + GIT_REPOSITORY https://github.com/google/re2.git + GIT_TAG 2025-11-05 # Use a stable version tag or 'main' +) +FetchContent_MakeAvailable(re2) + + find_package(Boost CONFIG) find_package(nlohmann_json CONFIG) + add_executable(StartUpRunner main.cpp appinfo.h appinfo.cpp executionmanager.h executionmanager.cpp log.h log.cpp) -target_link_libraries(StartUpRunner PRIVATE nlohmann_json::nlohmann_json) +target_link_libraries(StartUpRunner PRIVATE nlohmann_json::nlohmann_json re2::re2) diff --git a/appinfo.cpp b/appinfo.cpp index 8e25163..286a928 100644 --- a/appinfo.cpp +++ b/appinfo.cpp @@ -2,6 +2,7 @@ #include #include +#include AppInfo::~AppInfo() { std::cout << "Module: " << getModuleName() << " finished ops." << std::endl; @@ -57,6 +58,11 @@ void AppInfo::setRunAs(std::string run_as) this->run_as = run_as; } +void AppInfo::setScheduled(bool val) +{ + this->scheduled = val; +} + void AppInfo::setFinished() { this->state = State::exited; @@ -67,6 +73,11 @@ void AppInfo::setRunning() this->state = State::running; } +void AppInfo::addToPreviousExecutionList(pid_t new_process) +{ + this->previous_execution_list.push_back(new_process); +} + void AppInfo::insertParent(std::shared_ptr parent, State state) { this->parents[parent] = state; @@ -115,9 +126,10 @@ std::string AppInfo::getAfterString() const for(auto &[key,value]: after) { ss << key << "[" << value << "], "; } - std::string after = ss.str(); - after = after.substr(0,after.size() - 2); - return after; + std::string after_str = ss.str(); + if(after_str.size() >= 2) + after_str = after_str.substr(0, after_str.size() - 2); + return after_str; } std::string AppInfo::getAfterKeys() const @@ -126,9 +138,15 @@ std::string AppInfo::getAfterKeys() const for(auto &[key,value]: after) { ss << key << ", "; } - std::string after = ss.str(); - after = after.substr(0,after.size() - 2); - return after; + std::string after_str = ss.str(); + if(after_str.size() >= 2) + after_str = after_str.substr(0, after_str.size() - 2); + return after_str; +} + +std::string AppInfo::getStringRunMode() const +{ + return runmodeToString(); } AppInfo::RunMode AppInfo::getRunMode() const @@ -152,6 +170,9 @@ void AppInfo::copyFromData(AppInfo::AppData &other) } this->after = other.after; this->env = other.env; + this->run_at = other.run_at; + this->run_every = other.run_every; + this->run_on_start = other.run_on_start; setRunMode(other.run_mode); setRunAs(other.run_as); } @@ -171,6 +192,26 @@ bool AppInfo::getCreateLog() return create_log; } +bool AppInfo::getScheduled() +{ + return this->scheduled; +} + +bool AppInfo::getRunOnStart() +{ + return this->run_on_start; +} + +std::vector AppInfo::getPreviousExecutionList() const +{ + return this->previous_execution_list; +} + +std::chrono::system_clock::time_point AppInfo::getNextTimeToRun() const +{ + return this->next_time_to_run; +} + std::deque AppInfo::getArgs() const { return args; @@ -244,9 +285,371 @@ void AppInfo::printInfo(std::ostream &out) out << "Environment: " << env << std::endl; out << "Create Log: " << create_log << std::endl; out << "Run as: " << run_as << std::endl; + out << "Run on start: " << run_on_start << std::endl; + const auto* zone = std::chrono::current_zone(); + auto local_now = zone->to_local(this->next_time_to_run); + auto local_now_string = std::format("{:%F %T}", local_now); + if((run_at.size() != 0 || run_every.size() != 0)) { + out << "Next run scheduled at: " << local_now_string << std::endl; + } else { + out << "Normal scheduled run." << std::endl; + } out << "************************************End************************************" << std::endl; } +bool AppInfo::checkValidity() +{ + // The json is wrong is run on start is set when run_every and run_at are not set + if(this->run_every.size() == 0 && this->run_at.size() == 0 && this->run_on_start) + return false; + bool got_run_every = false; + bool got_run_at = false; + // The json is wrong if the format of run_every is wrong + std::chrono::system_clock::time_point every_time_point, at_time_point; + if(this->run_every.size() != 0) { + auto next_time = parseEveryTimeFormat(run_every); + if(next_time.has_value()) { + every_time_point = next_time.value(); + got_run_every = true; + } + else { + return false; + } + } + // The json is wrong if the format of run_at is wrong + if(this->run_at.size() != 0) { + auto next_time = parseAtTimeFormat(run_at); + if(next_time.has_value()) { + at_time_point = next_time.value(); + got_run_at = true; + } + else { + return false; + } + } + // Update next time to run + if(got_run_every && !got_run_at) { + scheduled = true; + next_time_to_run = every_time_point; + } + if(!got_run_every && got_run_at) { + scheduled = true; + next_time_to_run = at_time_point; + } + if(got_run_every && got_run_at) { + scheduled = true; + next_time_to_run = std::min(every_time_point, at_time_point); + } + return true; +} + +void AppInfo::updateNextTimeToRun() +{ + next_time_to_run_parse_failed = false; + bool got_run_every = false; + bool got_run_at = false; + std::chrono::system_clock::time_point every_time_point, at_time_point; + if(this->run_every.size() != 0) { + auto next_time = parseEveryTimeFormat(run_every); + if(next_time.has_value()) { + every_time_point = next_time.value(); + got_run_every = true; + } + else { + next_time_to_run_parse_failed = true; + } + } + if(this->run_at.size() != 0) { + auto next_time = parseAtTimeFormat(run_at); + if(next_time.has_value()) { + at_time_point = next_time.value(); + got_run_at = true; + } + else { + next_time_to_run_parse_failed = true; + } + } + if(next_time_to_run_parse_failed) + return; + if(got_run_every && !got_run_at) + next_time_to_run = every_time_point; + if(!got_run_every && got_run_at) + next_time_to_run = at_time_point; + if(got_run_every && got_run_at) + next_time_to_run = std::min(every_time_point, at_time_point); + scheduled = true; +} + +std::optional AppInfo::parseAtTimeFormat(std::string run_every) +{ + /* format should be like: + * "[time] Daily": for running daily at that time + * "[day] [time] Weekly": for running every week on those days at that time. + * "[day_month] [time] Monthly": for running every month on those days at that time. + * [time] format is 24Hrs with [HH:MM:SS] + * [day] format is any of [Mon, Tue, Wed, Thu, Fri, Sat, Sun] + * [day_month] format is 1-31. + * For Weekly, [day] can be a list of days seperated by commas. + * For Monthly, [day_month] can be a list of days seperated by commas. + * For Any, [time] can be a list of timestamps seperated by commas. + */ + + auto timeParseDaily = [this](std::string temp_run_every, std::chrono::system_clock::time_point now) -> std::optional { + using timepoint = std::chrono::system_clock::time_point; + std::vector possible_time_points; + std::vector time_list = tokenize(temp_run_every, ","); + const auto* zone = std::chrono::current_zone(); + auto local_now = zone->to_local(now); + std::string date_part = std::format("{:%F}", local_now); + + for(auto &time: time_list) { + std::string string_time_next = date_part + " " + time; + std::chrono::local_seconds local_next; + std::istringstream is{string_time_next}; + is >> std::chrono::parse("%F %T", local_next); + if(is.fail()) { + return std::nullopt; + } + if(local_now >= local_next) { + local_next = local_next + std::chrono::days(1); + } + timepoint next_time_point = zone->to_sys(local_next, std::chrono::choose::earliest); + possible_time_points.push_back(next_time_point); + } + if(possible_time_points.size() == 0) { + return std::nullopt; + } + return *std::ranges::min_element(possible_time_points); + }; + + auto timeParseWeekly = [this](std::string temp_run_every, std::string week_part, std::chrono::system_clock::time_point now) -> std::optional { + using timepoint = std::chrono::system_clock::time_point; + std::vector possible_time_points; + std::vector time_list = tokenize(temp_run_every, ","); + std::vector week_list = tokenize(week_part, ","); + const auto* zone = std::chrono::current_zone(); + auto local_now = zone->to_local(now); + int weekday_of_local_now = integer(std::format("{:%w}", local_now)); + auto getDayOfWeekInteger = [this](std::string day) -> std::optional { + if(tolower(day) == "sun") { + return 0; + } + else if(tolower(day) == "mon") { + return 1; + } + else if(tolower(day) == "tue") { + return 2; + } + else if(tolower(day) == "wed") { + return 3; + } + else if(tolower(day) == "thu") { + return 4; + } + else if(tolower(day) == "fri") { + return 5; + } + else if(tolower(day) == "sat") { + return 6; + } + else { + return std::nullopt; + } + }; + for(auto &day_of_week: week_list) { + auto temp_local_now = local_now; + auto day_of_week_integer_ret = getDayOfWeekInteger(day_of_week); + if(!day_of_week_integer_ret.has_value()) { + return std::nullopt; + } + int day_of_week_integer = day_of_week_integer_ret.value(); + if(day_of_week_integer != weekday_of_local_now) { + temp_local_now = temp_local_now + std::chrono::days( (day_of_week_integer - weekday_of_local_now + 7) % 7 ); + } + std::string date_part = std::format("{:%F}", temp_local_now); + for(auto &time: time_list) { + std::string string_time_next = date_part + " " + time; + std::chrono::local_seconds local_next; + std::istringstream is{string_time_next}; + is >> std::chrono::parse("%F %T", local_next); + if(is.fail()) { + return std::nullopt; + } + if(temp_local_now >= local_next && day_of_week_integer == weekday_of_local_now) { + local_next = local_next + std::chrono::days(7); + } + timepoint next_time_point = zone->to_sys(local_next, std::chrono::choose::earliest); + possible_time_points.push_back(next_time_point); + } + } + if(possible_time_points.size() == 0) { + return std::nullopt; + } + return *std::ranges::min_element(possible_time_points); + }; + + auto timeParseMonthly = [this](std::string temp_run_every, std::string month_part, std::chrono::system_clock::time_point now) -> std::optional { + using timepoint = std::chrono::system_clock::time_point; + std::vector possible_time_points; + std::vector time_list = tokenize(temp_run_every, ","); + std::vector month_days_list = tokenize(month_part, ","); + + const auto* zone = std::chrono::current_zone(); + auto local_now = zone->to_local(now); + std::chrono::year_month_day current_ymd{std::chrono::floor(local_now)}; + for(auto &day_of_month: month_days_list) { + int day_int = integer(day_of_month); + if (day_int < 1 || day_int > 31) { + return std::nullopt; + } + std::chrono::day target_day{static_cast(day_int)}; + std::chrono::year_month_day ymd = current_ymd.year() / current_ymd.month() / target_day; + if (!ymd.ok()) { + ymd = ymd.year() / ymd.month() / std::chrono::last; + } + for(auto &time: time_list) { + std::chrono::seconds time_of_day; + std::istringstream is{time}; + is >> std::chrono::parse("%T", time_of_day); + if (is.fail()) { + return std::nullopt; + } + auto local_next = std::chrono::local_days{ymd} + time_of_day; + if (local_next <= local_now) { + auto next_ym = current_ymd.year() / current_ymd.month() + std::chrono::months(1); + std::chrono::year_month_day ymd_next = next_ym / target_day; + if (!ymd_next.ok()) { + ymd_next = next_ym / std::chrono::last; + } + local_next = std::chrono::local_days{ymd_next} + time_of_day; + } + timepoint next_time_point = zone->to_sys(local_next, std::chrono::choose::earliest); + possible_time_points.push_back(next_time_point); + } + } + if(possible_time_points.size() == 0) { + return std::nullopt; + } + return *std::ranges::min_element(possible_time_points); + }; + + using timepoint = std::chrono::system_clock::time_point; + std::chrono::time_point now = std::chrono::system_clock::now(); + if(run_every.contains("Daily")) { + std::string temp_run_every = run_every.erase(run_every.find("Daily")); + temp_run_every = stringStrip(temp_run_every); + auto options = timeParseDaily(temp_run_every, now); + if(!options.has_value()) { + return std::nullopt; + } + return options.value(); + } + else if(run_every.contains("Weekly")) { + std::string temp_run_weekly = run_every.erase(run_every.find("Weekly")); + temp_run_weekly = stringStrip(temp_run_weekly); + std::string week_part; + std::string time_part; + if(!RE2::FullMatch(temp_run_weekly,R"(([a-zA-Z ,]+)([0-9: ,]+))", &week_part, &time_part)) { + return std::nullopt; + } + week_part = stringStrip(week_part); + time_part = stringStrip(time_part); + auto options = timeParseWeekly(time_part, week_part, now); + if(!options.has_value()) { + return std::nullopt; + } + return options.value(); + } + else if(run_every.contains("Monthly")) { + std::string temp_run_monthly = run_every.erase(run_every.find("Monthly")); + temp_run_monthly = stringStrip(temp_run_monthly); + std::string_view view_for_temp_run_monthy(temp_run_monthly); + std::string temp_time_part; + std::string time_part = ""; + re2::RE2 r(R"((\d\d:\d\d:\d\d[, ]?)+)"); + while (RE2::FindAndConsume(&view_for_temp_run_monthy, r, &temp_time_part)) { + time_part += temp_time_part; + } + RE2::GlobalReplace(&temp_run_monthly,R"((\d\d:\d\d:\d\d[, ]?)+)", ""); + std::string month_part = stringStrip(temp_run_monthly); time_part = stringStrip(time_part); + auto options = timeParseMonthly(time_part, month_part, now); + if(!options.has_value()) { + return std::nullopt; + } + return options.value(); + } + else { + return std::nullopt; + } +} + +std::optional AppInfo::parseEveryTimeFormat(std::string run_every) +{ + /* format should be like: + * [digits][unit] + * valid units are: + * s or S: seconds + * m: minutes + * h or H: hours + * d or D: days + * w or W: weeks + * M: months + * Example: 15s + * Example: 2h + */ + int duration = 0; + std::string unit; + if(!RE2::FullMatch(run_every,R"((\d+)([sSmMhHdDwW]))", &duration, &unit)) { + return std::nullopt; + } + if(duration <= 0) { + return std::nullopt; + } + if(unit.size() <= 0) { + return std::nullopt; + } + char ch = unit[0]; + std::chrono::system_clock::time_point now = std::chrono::system_clock::now(); + const auto* zone = std::chrono::current_zone(); + auto local_now = zone->to_local(now); + + switch(ch) { + case 's': + case 'S': { + return zone->to_sys(local_now + std::chrono::seconds(duration), std::chrono::choose::earliest); + } + case 'm': { + return zone->to_sys(local_now + std::chrono::minutes(duration), std::chrono::choose::earliest); + } + case 'h': + case 'H': { + return zone->to_sys(local_now + std::chrono::hours(duration), std::chrono::choose::earliest); + } + case 'd': + case 'D': { + return zone->to_sys(local_now + std::chrono::days(duration), std::chrono::choose::earliest); + } + case 'w': + case 'W': { + return zone->to_sys(local_now + std::chrono::weeks(duration), std::chrono::choose::earliest); + } + case 'M': { + auto today = std::chrono::floor(local_now); + auto time_of_day = local_now - today; + std::chrono::year_month_day ymd{today}; + ymd += std::chrono::months(duration); + if (!ymd.ok()) { + ymd = ymd.year() / ymd.month() / std::chrono::last; + } + auto next_local = std::chrono::local_days{ymd} + time_of_day; + return zone->to_sys(next_local, std::chrono::choose::earliest); + } + default: { + return std::nullopt; + } + } +} + std::string AppInfo::runmodeToString() const { switch(this->mode) { @@ -259,6 +662,67 @@ std::string AppInfo::runmodeToString() const } } +std::vector AppInfo::tokenize(std::string main_string, std::string sub_string) +{ + std::vector tokens; + size_t pos = 0; + while ((pos = main_string.find(sub_string)) != std::string::npos) { + std::string token = stringStrip(main_string.substr(0, pos)); + if (!token.empty()) { + tokens.push_back(token); + } + main_string.erase(0, pos + sub_string.length()); + } + std::string last_token = stringStrip(main_string); + if (!last_token.empty()) { + tokens.push_back(last_token); + } + return tokens; +} + +std::string AppInfo::stringStrip(const std::string str) +{ + std::string ch; + bool front_found = false; + for(int i=0; i> i) || !(ss >> std::ws).eof()) + return -1; + return i; +} + +std::string AppInfo::tolower(const std::string str) const +{ + std::string s; + std::transform(str.begin(), str.end(), std::back_inserter(s), [](unsigned char c){ + return static_cast(std::tolower(c)); + }); + return s; +} + std::string AppInfo::stateToString() const { return stateToString(this->state); @@ -309,6 +773,9 @@ void AppInfo::AppData::printInfo(std::ostream &out) out << "Environment: " << env << std::endl; out << "Create Log: " << AppData::create_log << std::endl; out << "Run as: " << run_as << std::endl; + out << "Run on start: " << run_on_start << std::endl; + out << "Run At: " << run_at << std::endl; + out << "Run Every: " << run_every << std::endl; out << "************************************End************************************" << std::endl; } @@ -318,9 +785,10 @@ std::string AppInfo::AppData::getAfterString() const for(auto &[key,value]: after) { ss << key << "[" << value << "], "; } - std::string after = ss.str(); - after = after.substr(0,after.size() - 2); - return after; + std::string after_str = ss.str(); + if(after_str.size() >= 2) + after_str = after_str.substr(0, after_str.size() - 2); + return after_str; } std::string AppInfo::AppData::getAfterKeys() const @@ -329,7 +797,8 @@ std::string AppInfo::AppData::getAfterKeys() const for(auto &[key,value]: after) { ss << key << ", "; } - std::string after = ss.str(); - after = after.substr(0,after.size() - 2); - return after; + std::string after_str = ss.str(); + if(after_str.size() >= 2) + after_str = after_str.substr(0, after_str.size() - 2); + return after_str; } diff --git a/appinfo.h b/appinfo.h index 040b3d8..50f5f10 100644 --- a/appinfo.h +++ b/appinfo.h @@ -8,6 +8,8 @@ #include #include #include +#include +#include class AppInfo { @@ -32,6 +34,9 @@ public: std::string run_mode; std::string run_as; bool create_log = false; + std::string run_at; + std::string run_every; + bool run_on_start = false; void printInfo(std::ostream &out); std::string getAfterString() const; std::string getAfterKeys() const; @@ -43,12 +48,19 @@ private: std::deque args; std::string run_as; std::string working_dir; + std::string run_at; + std::string run_every; + bool run_on_start = false; string_map after; string_map env; bool create_log = false; std::unordered_map, State> parents; std::atomic state{State::initial}; RunMode mode = RunMode::script; + std::chrono::system_clock::time_point next_time_to_run; + bool next_time_to_run_parse_failed = false; + bool scheduled = false; + std::vector previous_execution_list; public: void setProgramName(std::string name); void setModuleName(std::string name); @@ -59,8 +71,10 @@ public: void setCreateLog(bool create_log); void setRunMode(std::string mode); void setRunAs(std::string run_as); + void setScheduled(bool val); void setFinished(); void setRunning(); + void addToPreviousExecutionList(pid_t new_process); void insertParent(std::shared_ptr parent, State state); bool isParentDependencyCleared(std::osyncstream &out) const; std::string getModuleName() const; @@ -73,10 +87,15 @@ public: std::string getAfterString() const; std::string getAfterKeys() const; RunMode getRunMode() const; + std::string getStringRunMode() const; State getState() const; std::unordered_map, State> getParents() const; void copyFromData(AppInfo::AppData &other); bool getCreateLog(); + bool getScheduled(); + bool getRunOnStart(); + std::vector getPreviousExecutionList() const; + std::chrono::system_clock::time_point getNextTimeToRun() const; AppInfo(); ~AppInfo(); AppInfo(const AppInfo &other) = delete; @@ -84,8 +103,16 @@ public: AppInfo &operator=(const AppInfo &other) = delete; bool operator==(const AppInfo &other); void printInfo(std::ostream& out); + bool checkValidity(); + void updateNextTimeToRun(); private: + std::optional parseAtTimeFormat(std::string run_every); + std::optional parseEveryTimeFormat(std::string run_every); std::string runmodeToString() const; + std::vector tokenize(std::string, std::string); + std::string stringStrip(const std::string str); + int integer(const std::string str); + std::string tolower(const std::string str) const; std::string stateToString() const; std::string stateToString(State s) const; }; diff --git a/executionmanager.cpp b/executionmanager.cpp index 2f6889f..ab3f672 100644 --- a/executionmanager.cpp +++ b/executionmanager.cpp @@ -46,44 +46,68 @@ void ExecutionManager::setJobs(std::vector > &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) { - for(int i=active_thread_count.load(); i < threads ;i++) { + // 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(); + + // Dispatch jobs from job_list to worker threads + while(active_thread_count.load() < threads) { + std::shared_ptr ptr; int val = 0; + size_t remaining_size = 0; { std::lock_guard 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)) { - log_file << "Main Thread: All jobs queued." << std::endl; 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 thread = std::make_shared([this, val](){ - std::shared_ptr ptr; - std::osyncstream os(log_file.rdbuf()); - size_t list_size = 0; - { - std::lock_guard 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(); + worker->thread = std::thread([this, val, ptr, remaining_size, worker](){ + struct ThreadGuard { + std::atomic& count; + std::atomic& 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)) { @@ -178,13 +202,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 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,29 +230,68 @@ 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; - active_thread_count.fetch_sub(1); + << module_name << std::endl; }); - 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 lock_get(this->job_list_mutex); - if(job_list.empty()) { - 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; + 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; } + +bool ExecutionManager::checkScheduledTaskIsReady(std::chrono::system_clock::time_point next_time) +{ + return (next_time <= std::chrono::system_clock::now()); +} + +void ExecutionManager::seperateScheduledTasks() +{ + std::lock_guard 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(*task); + task = job_list.erase(task); + } else { + ++task; + } + } +} + +void ExecutionManager::pollScheduledTasks() +{ + auto current_poll_time = std::chrono::system_clock::now(); + if(std::chrono::duration_cast(current_poll_time - last_poll_time).count() < polling_interval) + return; + last_poll_time = current_poll_time; + { + std::lock_guard lock_get(this->job_list_mutex); + for(auto task = scheduled_list.begin(); task != scheduled_list.end(); ) { + if(checkScheduledTaskIsReady((*task)->getNextTimeToRun())) { + 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); + } else { + ++task; + } + } + } +} diff --git a/executionmanager.h b/executionmanager.h index d95e3dd..c90b905 100644 --- a/executionmanager.h +++ b/executionmanager.h @@ -10,17 +10,23 @@ class ExecutionManager { + struct WorkerThread { + std::thread thread; + std::atomic done{false}; + }; bool finished = false; int threads; int index_count = 0; - int initial_list_size = 0; long long int log_file_max_size = 0; std::string log_dir; std::mutex job_list_mutex; std::atomic state_change = 0; std::atomic active_thread_count = 0; - std::vector> thread_list; + std::vector> thread_list; std::vector> job_list; + std::vector> scheduled_list; + std::chrono::system_clock::time_point last_poll_time; + int polling_interval = 1; Log log_file; public: ExecutionManager(int threads=16); @@ -28,7 +34,12 @@ public: void setLogFileMaxSize(long long int size); void setLogDirectoy(std::string log_dir); void setJobs(std::vector>& job_list); + void printActiveJobs(); void run(); +private: + bool checkScheduledTaskIsReady(std::chrono::system_clock::time_point next_time); + void seperateScheduledTasks(); + void pollScheduledTasks(); }; #endif // EXECUTIONMANAGER_H diff --git a/main.cpp b/main.cpp index 3e88be3..8caf0c4 100644 --- a/main.cpp +++ b/main.cpp @@ -89,55 +89,81 @@ std::string strip(std::string& t) { } void printHelp() { - std::cout << "Usage: StartUpRunner [OPTIONS]" << std::endl; - std::cout << "OPTIONS:" << std::endl; - std::cout << "\t--help|-h : Prints this screen." << std::endl; - std::cout << "\t--version|-v : Prints version info." << std::endl; - std::cout << "\t--threads|-t [T] : Sets number of threads to [T]." << std::endl; - std::cout << "\t--log_dir|-ld [Dir] : Sets log directory to [Dir]." << std::endl; - std::cout << "\t--defaults|-d : Prints the default values used if no arguments are provided." << std::endl; - std::cout << "\t--json|-j [FileName]: Set Json Startup List to FileName." << std::endl; - std::cout << "\t--test|-T [FileName]: Check the json file for errors." << std::endl; - std::cout << "For a detailed info about the json startup file use --help json." << std::endl; + std::cout << R"(Usage: StartUpRunner [OPTIONS] +OPTIONS: + --help|-h : Prints this screen. + --version|-v : Prints version info. + --threads|-t [T] : Sets number of threads to [T]. + --log_dir|-ld [Dir] : Sets log directory to [Dir]. + --defaults|-d : Prints the default values used if no arguments are provided. + --json|-j [FileName]: Set Json Startup List to FileName. + --test|-T [FileName]: Check the json file for errors. +For a detailed info about the json startup file use --help json.)" << std::endl; } void printVersion() { - std::cout << "StartUpRunner: Version-0.0.2 Pre-Alpha Pre-BigBang Era Edition." << std::endl; + std::cout << "StartUpRunner: Version-0.0.3 Pre-Alpha Pre-BigBang Era Edition." << std::endl; } void printDefaults() { std::cout << "StartUpRunner Default Values: " << std::endl; std::cout << "Json file : " << options.json_file_path << std::endl; std::cout << "Logs directory : " << options.log_directory << std::endl; - std::cout << "Log file max size: " << options.threads << std::endl; + std::cout << "Log file max size: " << options.log_file_max_size << std::endl; std::cout << "Threads : " << options.threads << std::endl; } void printJsonHelp() { - std::cout << "Json StartUp List file format: " << std::endl; - std::cout << "Top level is an Array of Apps." << std::endl; - std::cout << "Each App Entry contains: " << std::endl; - std::cout << "\t[REQUIRED] name[string] : Name of the module." << std::endl; - std::cout << "\t[REQUIRED] app_name[string] : Path to the program with its name." << std::endl; - std::cout << "\t[OPTIONAL] app_args[list[string]] : List of arguments to the program." << std::endl; - std::cout << "\t[REQUIRED] run_mode[string] : Mode can be daemon or script." << std::endl; - std::cout << "\t\tMode daemon : The app forks the program and exits without waiting." << std::endl; - std::cout << "\t\tMode script : The app forks the program and waits for it to finish execution." << std::endl; - std::cout << "\t\t If no run_mode is given it is presumed to be script." << std::endl; - std::cout << "\t[OPTIONAL] after[{name,state}] : List containing names and state of modules." << std::endl; - std::cout << "\t\tname[string] : Name of the other module." << std::endl; - std::cout << "\t\tstate[string]: State of the other module." << std::endl; - std::cout << "\t\t\tState exited : Specified module must have finished execution." << std::endl; - std::cout << "\t\t\tState running : Specified module must be running." << std::endl; - std::cout << "\t\tThis module will be launched after the module found\n\t\t" - "in this list using name is at the state specified as state." << std::endl; - std::cout << "\t[OPTIONAL] env{} : Additional enviroment variables." << std::endl; - std::cout << "\t\tEvery key in this object will be used as environment name,\n\t\tand every corresponding value as its value." << std::endl; - std::cout << "\t[OPTIONAL] working_dir[string] : Working directory of the program." << std::endl; - std::cout << "\t[OPTIONAL] create_log[boolean] : Create a log file at logs path." << std::endl; - std::cout << "\t[OPTIONAL] run_as[string] : Run as a particular user." << std::endl; - std::cout << std::endl << "Example startup_apps.json: "; - std::cout << R"( + std::cout << R"(Json StartUp List file format: +Top level is an Array of Apps. +Each App Entry contains: + [REQUIRED] name[string] : Name of the module. + [REQUIRED] app_name[string] : Path to the program with its name. + [REQUIRED] run_mode[string] : Mode can be daemon or script. + [OPTIONAL] app_args[list[string]] : List of arguments to the program. + Mode daemon : The app forks the program and exits without waiting. + Mode script : The app forks the program and waits for it to finish execution. + If no run_mode is given it is presumed to be script. + [OPTIONAL] after[{name,state}] : List containing names and state of modules. + name[string] : Name of the other module. + state[string]: State of the other module. + State exited : Specified module must have finished execution. + State running : Specified module must be running. + This module will be launched after the module foundin this list using name is at the state specified as state. + [OPTIONAL] env{key: value} : Additional enviroment variables. + Every key in this object will be used as environment name, + and every corresponding value as its value. + [OPTIONAL] working_dir[string] : Working directory of the program. + [OPTIONAL] create_log[boolean] : Create a log file at logs path. + [OPTIONAL] run_as[string] : Run as a particular user. + [OPTIONAL] run_every[string] : Run task at every specified interval. + format should be like: + [digits][unit] + valid units are: + s or S: seconds + m: minutes + h or H: hours + d or D: days + w or W: weeks + M: months + Example: 15s + Example: 2h + [OPTIONAL] run_at[string] : Run task at every specified time. + Format of string should be like: + "[time] Daily": for running daily at that time + "[day] [time] Weekly": for running every week on those days at that time. + "[day_month] [time] Monthly": for running every month on those days at that time. + [time] format is 24Hrs with [HH:MM:SS] + [day] format is any of [Mon, Tue, Wed, Thu, Fri, Sat, Sun] + [day_month] format is 1-31. + For Weekly, [day] can be a list of days seperated by commas. + For Monthly, [day_month] can be a list of days seperated by commas. + For Any, [time] can be a list of timestamps seperated by commas. + [time] Daily: for running daily at that time. + [day] [time] Weekly: for running every week on those days at that time. + [OPTIONAL] run_on_start[boolean] : Run on initial startup as well [Only valid for scheduled tasks]. + +Example startup_apps.json: [ { "name": "LS 0", @@ -156,10 +182,24 @@ void printJsonHelp() { "after": [{"module": "LS 0", "state": "exited"}], "env": {"PATH": "/home/:/usr/bin/", "NAME": "ONE"}, "working_dir": "/home/etc", - "run_as": "username" + "run_as": "username", + "run_every": "10s", + "run_on_start": true + }, + { + "name": "LS 2", + "app_name": "/bin/ls", + "app_args": ["-lh"], + "run_mode": "script", + "create_log": true, + "after": [{"module": "LS 0", "state": "exited"}], + "working_dir": "/home/etc/Documents", + "run_on_start": false, + "run_at": "1,15,30 01:30:00 Monthly" } ] -)" << std::endl; +)"; + } int contains(std::vector &list, std::string l, std::string v) { @@ -206,7 +246,7 @@ void parseArgs(int c, char *v[]) { int i = contains(arg_list, "--json", "-j"); if(i >= 0) { int j = i + 1; - if(j > arg_list.size()) { + if(j >= arg_list.size()) { printHelp(); exit(-1); } @@ -217,7 +257,7 @@ void parseArgs(int c, char *v[]) { i = contains(arg_list, "--log_dir", "-ld"); if(i >= 0) { int j = i + 1; - if(j > arg_list.size()) { + if(j >= arg_list.size()) { printHelp(); exit(-1); } @@ -228,7 +268,7 @@ void parseArgs(int c, char *v[]) { i = contains(arg_list, "--threads", "-t"); if(i >= 0) { int j = i + 1; - if(j > arg_list.size()) { + if(j >= arg_list.size()) { printHelp(); exit(-1); } @@ -238,29 +278,24 @@ void parseArgs(int c, char *v[]) { } i = contains(arg_list, "--test", "-T"); if(i >= 0) { - int j = i + 1; - if(j > arg_list.size()) { - printHelp(); - exit(-1); - } options.check_for_errors = true; + int j = i + 1; + if(j < arg_list.size() && arg_list[j].rfind("-", 0) != 0) { + options.json_file_path = arg_list[j]; + } std::cout << "Testing json file." << std::endl; contains_any_known_arg = true; } i = contains(arg_list, "--help", "-h"); if(i >= 0) { int j = i + 1; - if(j > arg_list.size()) { - printHelp(); - exit(-1); - } - if(arg_list[j] == "json") { + if(j < arg_list.size() && arg_list[j] == "json") { printJsonHelp(); exit(0); } else { printHelp(); - exit(-1); + exit(0); } contains_any_known_arg = true; } @@ -299,11 +334,14 @@ std::vector string_split(std::string s, const std::string& delimite size_t pos = 0; std::string token; while ((pos = s.find(delimiter)) != std::string::npos) { - token = s.substr(0, pos); - tokens.push_back(token); + token = string_strip(s.substr(0, pos)); + if(!token.empty()) + tokens.push_back(token); s.erase(0, pos + delimiter.length()); } - tokens.push_back(string_strip(s)); + std::string last = string_strip(s); + if(!last.empty()) + tokens.push_back(last); return tokens; } @@ -396,7 +434,7 @@ std::optional checkAppList(std::deque ©_list, std::ve } } if(!found) { - out << "Depenedency not found" << std::endl; + out << "Dependency not found" << std::endl; } } after_list.push_back(std::move(app_info)); @@ -423,14 +461,41 @@ std::optional checkAppList(std::deque ©_list, std::ve } } } + bool all_valid = true; for(auto &i: after_list) { + bool run_validity = i->checkValidity(); + i->printInfo(out); for(auto &[j, state]: i->getParents()) { if(state == AppInfo::State::exited && j->getRunMode() == AppInfo::RunMode::daemon) { out << "[WARN]: " << i->getModuleName() << - " is waiting for a daemon: " << j->getModuleName() << " to start exeuction." << + " is waiting for a daemon: " << j->getModuleName() << " to start execution." << std::endl; } } + if(!run_validity) { + out << "[ERR]: " << i->getModuleName() << + " incorrect format for scheduled run. Please see help page." << std::endl; + all_valid = false; + } else if((i->getScheduled()) && (!i->getRunOnStart())) { + const auto* zone = std::chrono::current_zone(); + auto local_now = zone->to_local(i->getNextTimeToRun()); + out << "Module: " << i->getModuleName() << + " is scheduled for execution at: " << + std::format("{:%F %T}", local_now) << std::endl; + } else if((i->getScheduled()) && (i->getRunOnStart())) { + const auto* zone = std::chrono::current_zone(); + auto local_now = zone->to_local(i->getNextTimeToRun()); + out << "Module: " << i->getModuleName() << + " will run on init and is also scheduled for execution at: " << + std::format("{:%F %T}", local_now) << std::endl; + } + else { + out << "Module: " << i->getModuleName() << + " is to be launched on init." << std::endl; + } + } + if(!all_valid) { + return std::nullopt; } return 0; } @@ -541,6 +606,13 @@ int main(int argc, char *argv[]) { app_info.create_log = object["create_log"].get(); if(object.find("run_as") != object.end()) app_info.run_as = object["run_as"].get(); + if(object.find("run_at") != object.end()) + app_info.run_at = object["run_at"].get(); + if(object.find("run_every") != object.end()) + app_info.run_every = object["run_every"].get(); + if(object.find("run_on_start") != object.end()) + app_info.run_on_start = object["run_on_start"].get(); + app_info.env = env_list; app_info.after = after_list; app_info.module_name = object["name"].get();