Adding executor description for verbose output, adding output directory for build artifacts
This commit is contained in:
@@ -10,6 +10,10 @@ set(CMAKE_CXX_STANDARD_REQUIRED True)
|
|||||||
set(CMAKE_EXPORT_COMPILE_COMMANDS True)
|
set(CMAKE_EXPORT_COMPILE_COMMANDS True)
|
||||||
set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -g -Wall -Werror")
|
set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -g -Wall -Werror")
|
||||||
|
|
||||||
|
set(CMAKE_ARCHIVE_OUTPUT_DIRECTORY ${CMAKE_BINARY_DIR}/lib)
|
||||||
|
set(CMAKE_LIBRARY_OUTPUT_DIRECTORY ${CMAKE_BINARY_DIR}/lib)
|
||||||
|
set(CMAKE_RUNTIME_OUTPUT_DIRECTORY ${CMAKE_BINARY_DIR}/bin)
|
||||||
|
|
||||||
if(CMAKE_BUILD_TYPE MATCHES "Debug")
|
if(CMAKE_BUILD_TYPE MATCHES "Debug")
|
||||||
set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -fno-omit-frame-pointer")
|
set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -fno-omit-frame-pointer")
|
||||||
# set(TSAN_OPTIONS "suppressions=${CMAKE_CURRENT_DIR}/tests/tsan.supp")
|
# set(TSAN_OPTIONS "suppressions=${CMAKE_CURRENT_DIR}/tests/tsan.supp")
|
||||||
|
|||||||
@@ -270,11 +270,15 @@ int main(int argc, char **argv)
|
|||||||
doc.SetObject();
|
doc.SetObject();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
auto logger = loggerFactory(doc);
|
||||||
|
auto executor = executorFactory(doc);
|
||||||
|
|
||||||
if (verbose) {
|
if (verbose) {
|
||||||
std::cout << "Server running at http://" << listenIP << ':' << listenPort
|
std::cout << "Server running at http://" << listenIP << ':' << listenPort
|
||||||
<< std::endl
|
<< std::endl
|
||||||
<< "Max DAG Processing: " << dagThreads << std::endl
|
<< "Max DAG Processing: " << dagThreads << std::endl
|
||||||
<< "Max Web Clients: " << webThreads << std::endl
|
<< "Max Web Clients: " << webThreads << std::endl
|
||||||
|
<< "Executor: " << executor->description() << std::endl
|
||||||
<< std::endl
|
<< std::endl
|
||||||
<< "Ctrl-C to exit" << std::endl;
|
<< "Ctrl-C to exit" << std::endl;
|
||||||
}
|
}
|
||||||
@@ -283,9 +287,6 @@ int main(int argc, char **argv)
|
|||||||
daemonize();
|
daemonize();
|
||||||
}
|
}
|
||||||
|
|
||||||
auto logger = loggerFactory(doc);
|
|
||||||
auto executor = executorFactory(doc);
|
|
||||||
|
|
||||||
Pistache::Address listenSpec(listenIP, listenPort);
|
Pistache::Address listenSpec(listenIP, listenPort);
|
||||||
|
|
||||||
daggy::daggyd::Server server(listenSpec, *logger, *executor, dagThreads);
|
daggy::daggyd::Server server(listenSpec, *logger, *executor, dagThreads);
|
||||||
|
|||||||
@@ -169,6 +169,11 @@ namespace daggy {
|
|||||||
cv_.notify_one();
|
cv_.notify_one();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
size_t size() const
|
||||||
|
{
|
||||||
|
return workers_.size();
|
||||||
|
}
|
||||||
|
|
||||||
private:
|
private:
|
||||||
// need to keep track of threads, so we can join them
|
// need to keep track of threads, so we can join them
|
||||||
std::vector<std::thread> workers_;
|
std::vector<std::thread> workers_;
|
||||||
|
|||||||
@@ -51,6 +51,8 @@ namespace daggy::executors::task {
|
|||||||
|
|
||||||
bool stop(DAGRunID runID, const std::string &taskName) override;
|
bool stop(DAGRunID runID, const std::string &taskName) override;
|
||||||
|
|
||||||
|
std::string description() const;
|
||||||
|
|
||||||
void addRunner(const std::string &url);
|
void addRunner(const std::string &url);
|
||||||
|
|
||||||
private:
|
private:
|
||||||
|
|||||||
@@ -31,6 +31,8 @@ namespace daggy::executors::task {
|
|||||||
|
|
||||||
bool stop(DAGRunID runID, const std::string &taskName) override;
|
bool stop(DAGRunID runID, const std::string &taskName) override;
|
||||||
|
|
||||||
|
std::string description() const override;
|
||||||
|
|
||||||
private:
|
private:
|
||||||
ThreadPool tp_;
|
ThreadPool tp_;
|
||||||
std::mutex taskControlsGuard_;
|
std::mutex taskControlsGuard_;
|
||||||
|
|||||||
@@ -21,5 +21,7 @@ namespace daggy::executors::task {
|
|||||||
const Task &task) override;
|
const Task &task) override;
|
||||||
|
|
||||||
bool stop(DAGRunID runID, const std::string &taskName) override;
|
bool stop(DAGRunID runID, const std::string &taskName) override;
|
||||||
|
|
||||||
|
std::string description() const;
|
||||||
};
|
};
|
||||||
} // namespace daggy::executors::task
|
} // namespace daggy::executors::task
|
||||||
|
|||||||
@@ -25,6 +25,8 @@ namespace daggy::executors::task {
|
|||||||
|
|
||||||
bool stop(DAGRunID runID, const std::string &taskName) override;
|
bool stop(DAGRunID runID, const std::string &taskName) override;
|
||||||
|
|
||||||
|
std::string description() const override;
|
||||||
|
|
||||||
private:
|
private:
|
||||||
struct Job
|
struct Job
|
||||||
{
|
{
|
||||||
|
|||||||
@@ -33,5 +33,8 @@ namespace daggy::executors::task {
|
|||||||
|
|
||||||
// Kill a currently executing task. This will resolve the future.
|
// Kill a currently executing task. This will resolve the future.
|
||||||
virtual bool stop(DAGRunID runID, const std::string &taskName) = 0;
|
virtual bool stop(DAGRunID runID, const std::string &taskName) = 0;
|
||||||
|
|
||||||
|
// Report a description of the executor
|
||||||
|
virtual std::string description() const = 0;
|
||||||
};
|
};
|
||||||
} // namespace daggy::executors::task
|
} // namespace daggy::executors::task
|
||||||
|
|||||||
@@ -88,6 +88,25 @@ DaggyRunnerTaskExecutor::~DaggyRunnerTaskExecutor()
|
|||||||
monitorWorker_.join();
|
monitorWorker_.join();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
std::string DaggyRunnerTaskExecutor::description() const
|
||||||
|
{
|
||||||
|
std::stringstream ss;
|
||||||
|
ss << "DaggyRunnerTaskExecutor running with " << runners_.size()
|
||||||
|
<< " runners: [";
|
||||||
|
bool first = true;
|
||||||
|
for (const auto &[runner, _] : runners_) {
|
||||||
|
if (first) {
|
||||||
|
first = false;
|
||||||
|
}
|
||||||
|
else {
|
||||||
|
ss << ", ";
|
||||||
|
}
|
||||||
|
ss << runner;
|
||||||
|
}
|
||||||
|
ss << "]";
|
||||||
|
return ss.str();
|
||||||
|
}
|
||||||
|
|
||||||
// Validates the job to ensure that all required values are set and are of
|
// Validates the job to ensure that all required values are set and are of
|
||||||
// the right type,
|
// the right type,
|
||||||
bool DaggyRunnerTaskExecutor::validateTaskParameters(const ConfigValues &job)
|
bool DaggyRunnerTaskExecutor::validateTaskParameters(const ConfigValues &job)
|
||||||
|
|||||||
@@ -72,6 +72,13 @@ ForkingTaskExecutor::~ForkingTaskExecutor()
|
|||||||
taskControls_.clear();
|
taskControls_.clear();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
std::string ForkingTaskExecutor::description() const
|
||||||
|
{
|
||||||
|
std::stringstream ss;
|
||||||
|
ss << "ForkingTaskExecutor with " << tp_.size() << " threads";
|
||||||
|
return ss.str();
|
||||||
|
}
|
||||||
|
|
||||||
bool ForkingTaskExecutor::stop(DAGRunID runID, const std::string &taskName)
|
bool ForkingTaskExecutor::stop(DAGRunID runID, const std::string &taskName)
|
||||||
{
|
{
|
||||||
std::string key = std::to_string(runID) + "_" + taskName;
|
std::string key = std::to_string(runID) + "_" + taskName;
|
||||||
|
|||||||
@@ -2,6 +2,12 @@
|
|||||||
#include <daggy/executors/task/NoopTaskExecutor.hpp>
|
#include <daggy/executors/task/NoopTaskExecutor.hpp>
|
||||||
|
|
||||||
namespace daggy::executors::task {
|
namespace daggy::executors::task {
|
||||||
|
|
||||||
|
std::string NoopTaskExecutor::description() const
|
||||||
|
{
|
||||||
|
return "NoopTaskExecutor";
|
||||||
|
}
|
||||||
|
|
||||||
std::future<daggy::AttemptRecord> NoopTaskExecutor::execute(
|
std::future<daggy::AttemptRecord> NoopTaskExecutor::execute(
|
||||||
DAGRunID runID, const std::string &taskName, const Task &task)
|
DAGRunID runID, const std::string &taskName, const Task &task)
|
||||||
{
|
{
|
||||||
|
|||||||
@@ -93,6 +93,11 @@ namespace daggy::executors::task {
|
|||||||
runningJobs_.clear();
|
runningJobs_.clear();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
std::string SlurmTaskExecutor::description() const
|
||||||
|
{
|
||||||
|
return "SlurmTaskExecutor";
|
||||||
|
}
|
||||||
|
|
||||||
// Validates the job to ensure that all required values are set and are of
|
// Validates the job to ensure that all required values are set and are of
|
||||||
// the right type,
|
// the right type,
|
||||||
bool SlurmTaskExecutor::validateTaskParameters(const ConfigValues &job)
|
bool SlurmTaskExecutor::validateTaskParameters(const ConfigValues &job)
|
||||||
|
|||||||
Reference in New Issue
Block a user