made threaded_run() private, everything should use parallel()

comments
This commit is contained in:
isanae
2020-12-03 17:45:35 -05:00
parent 01797282c4
commit 8272db8233
9 changed files with 332 additions and 177 deletions
+2 -2
View File
@@ -82,8 +82,8 @@ void list_command::dump(const std::vector<task*>& v, std::size_t indent) const
<< " - " << join(t->names(), ",")
<< "\n";
if (auto* ct=dynamic_cast<container_task*>(t))
dump(ct->children(), indent + 1);
if (auto* pt=dynamic_cast<parallel_tasks*>(t))
dump(pt->children(), indent + 1);
}
}
+10 -8
View File
@@ -7,13 +7,17 @@ namespace mob::tasks
static std::mutex g_super_mutex;
static std::atomic<bool> g_super_initialized = false;
std::string make_short_name(const std::string& name)
std::vector<std::string> make_names(std::vector<std::string> names)
{
const auto dash = name.find("-");
if (dash == std::string::npos)
return name;
auto short_name = names[0];
return name.substr(dash + 1);
const auto dash = short_name.find("-");
if (dash != std::string::npos)
short_name = short_name.substr(dash + 1);
names.insert(names.begin(), short_name);
return names;
}
@@ -28,10 +32,8 @@ modorganizer::modorganizer(std::vector<const char*> names, flags f)
}
modorganizer::modorganizer(std::vector<std::string> names, flags f)
: basic_task(make_short_name(names[0])), repo_(names[0]), flags_(f)
: basic_task(make_names(names)), repo_(names[0]), flags_(f)
{
for (auto&& n : names)
add_name(std::move(n));
}
std::string modorganizer::version()
+85 -95
View File
@@ -19,7 +19,6 @@ std::string to_string(task::clean c)
case task::clean::reextract: break;
case task::clean::reconfigure: break;
case task::clean::rebuild: break;
case task::clean::everything: break;
}
std::vector<std::string> v;
@@ -39,6 +38,26 @@ std::string to_string(task::clean c)
return join(v, "|");
}
task::clean make_clean_flags()
{
task::clean c = task::clean::nothing;
const auto g = conf().global();
if (g.redownload())
c |= task::clean::redownload;
if (g.reextract())
c |= task::clean::reextract;
if (g.reconfigure())
c |= task::clean::reconfigure;
if (g.rebuild())
c |= task::clean::rebuild;
return c;
}
struct task::thread_context
{
@@ -62,23 +81,29 @@ task::task(std::vector<std::string> names)
task_manager::instance().register_task(this);
}
task::~task()
{
try
{
join();
}
catch(bailed&)
{
// ignore
}
}
// anchor
task::~task() = default;
bool task::enabled() const
{
return conf().task(names()).get<bool>("enabled");
}
void task::do_clean(clean)
{
// no-op
}
void task::do_fetch()
{
// no-op
}
void task::do_build_and_install()
{
// no-op
}
const context& task::cx() const
{
static const context bad("?");
@@ -96,15 +121,6 @@ const context& task::cx() const
return bad;
}
void task::add_name(std::string s)
{
auto itor = std::find(names_.begin(), names_.end(), s);
if (itor != names_.end())
return;
names_.push_back(s);
}
const std::string& task::name() const
{
return names_[0];
@@ -226,22 +242,19 @@ void task::threaded_run(std::string thread_name, std::function<void ()> f)
}
}
void task::parallel(std::vector<std::pair<std::string, std::function<void ()>>> v)
void task::parallel(parallel_functions v, std::optional<std::size_t> threads)
{
std::vector<std::thread> ts;
thread_pool tp(threads);
for (auto&& [name, f] : v)
{
cx().trace(context::generic, "running in parallel: {}", name);
ts.push_back(start_thread([this, name, f]
tp.add([this, name, f]
{
threaded_run(name, f);
}));
});
}
for (auto&& t : ts)
t.join();
}
conf_task task::task_conf() const
@@ -249,10 +262,9 @@ conf_task task::task_conf() const
return conf().task(names());
}
git task::make_git(git::ops o) const
git task::make_git() const
{
if (o == git::clone_or_pull && task_conf().no_pull())
o = git::clone;
const auto o = task_conf().no_pull() ? git::clone : git::clone_or_pull;
git g(o);
@@ -279,6 +291,16 @@ std::string task::make_git_url(
return task_conf().git_url_prefix() + org + "/" + repo + ".git";
}
fs::path task::get_source_path() const
{
return {};
}
bool task::get_prebuilt() const
{
return false;
}
void task::run()
{
threaded_run(name(), [&]
@@ -292,13 +314,9 @@ void task::run()
cx().info(context::generic, "running task");
fetch();
join();
check_interrupted();
build_and_install();
join();
check_interrupted();
});
}
@@ -313,12 +331,6 @@ void task::interrupt()
t->interrupt();
}
void task::join()
{
if (thread_.joinable())
thread_.join();
}
void task::clean_task()
{
if (!conf().global().clean())
@@ -348,33 +360,27 @@ void task::fetch()
return;
}
thread_ = start_thread([&]
clean_task();
check_interrupted();
if (conf().global().fetch())
{
threaded_run(name(), [&]
cx().info(context::generic, "fetching");
do_fetch();
check_interrupted();
if (!get_source_path().empty())
{
clean_task();
check_interrupted();
cx().debug(context::generic, "patching");
if (conf().global().fetch())
{
cx().info(context::generic, "fetching");
do_fetch();
run_tool(patcher()
.task(name(), get_prebuilt())
.root(get_source_path()));
}
check_interrupted();
if (!get_source_path().empty())
{
cx().debug(context::generic, "patching");
run_tool(patcher()
.task(name(), get_prebuilt())
.root(get_source_path()));
}
check_interrupted();
}
});
});
check_interrupted();
}
}
void task::build_and_install()
@@ -389,38 +395,12 @@ void task::build_and_install()
return;
}
thread_ = start_thread([&]
{
threaded_run(name(), [&]
{
check_interrupted();
check_interrupted();
cx().info(context::generic, "build and install");
do_build_and_install();
cx().info(context::generic, "build and install");
do_build_and_install();
check_interrupted();
});
});
}
task::clean task::make_clean_flags() const
{
clean c = clean::nothing;
const auto g = conf().global();
if (g.redownload())
c |= clean::redownload;
if (g.reextract())
c |= clean::reextract;
if (g.reconfigure())
c |= clean::reconfigure;
if (g.rebuild())
c |= clean::rebuild;
return c;
check_interrupted();
}
void task::check_interrupted()
@@ -461,10 +441,20 @@ void task::run_tool_impl(tool* t)
parallel_tasks::parallel_tasks()
: container_task("parallel")
: task("parallel")
{
}
parallel_tasks::~parallel_tasks()
{
join();
}
bool parallel_tasks::enabled() const
{
return true;
}
void parallel_tasks::add_task(std::unique_ptr<task> t)
{
children_.push_back(std::move(t));
+210 -58
View File
@@ -1,35 +1,71 @@
#pragma once
#include "../utility.h"
#include "../core/context.h"
#include "../tools/tools.h"
namespace mob
{
class task;
class tool;
class conf_task;
class git;
// base class for all tasks, although tasks will actually inherit from
// basic_task<> below
//
class task
{
public:
// passed to do_clean(), can be any combination depending on the
// configuration (redownload, reextract, etc. in the ini)
//
enum class clean
{
// do_clean() never gets that
nothing = 0x00,
// whatever downloaded file that was cached must be deleted so it can
// be downloaded again; cached downloads normally inhibit downloads
redownload = 0x01,
// extracting or cloning must be fresh, so basically delete the whole
// source directory; both enums have the same value, but `reclone` makes
// more sense for tools that don't actually extract archives
reextract = 0x02,
reclone = reextract,
// building the task must run whatever configuration tool from scratch,
// such as deleting the vsbuild directory for cmake tasks
reconfigure = 0x04,
rebuild = 0x08,
everything = redownload+reextract+reconfigure+rebuild
// the task must be rebuilt from scratch, but without reconfiguration,
// such as `msbuild` with the `Clean` target
//
// some tasks don't have an equivalent for this and might just delete
// the whole source directory, such as openssl
rebuild = 0x08
};
task(const task&) = delete;
task& operator=(const task&) = delete;
// anchor
//
virtual ~task();
// whether this task is enabled, just checks conf().task(), but
// parallel_tasks overrides this below to always be true
//
virtual bool enabled() const;
// main task name
//
const std::string& name() const;
// all names for this task
//
const std::vector<std::string>& names() const;
// case insensitive, underscores and dashes are equivalent; gets converted
@@ -37,15 +73,34 @@ public:
//
bool name_matches(std::string_view pattern) const;
virtual fs::path get_source_path() const = 0;
virtual std::string get_version() const = 0;
virtual const bool get_prebuilt() const = 0;
// path to the source directory, something like prefix/build/7zip-xx or
// or prefix/build/modorganizer_super/uibase
//
// used for auto patching in fetch(), returns an empty path here
//
virtual fs::path get_source_path() const;
// whether this task should use the prebuilt version
//
// used for auto patching in fetch(), returns false here
//
virtual bool get_prebuilt() const;
// if the task is enabled, calls fetch() and build_and_install()
//
virtual void run();
// sets the interrupt flag on this task so it's picked up in run() and
// calls interrupt() on all tools currently running
//
virtual void interrupt();
virtual void join();
protected:
using parallel_functions =
std::vector<std::pair<std::string, std::function<void ()>>>;
template <class... Names>
task(std::string name, Names&&... names)
: task(std::vector<std::string>{name, std::forward<Names>(names)...})
@@ -54,15 +109,40 @@ protected:
task(std::vector<std::string> names);
const context& cx() const;
void add_name(std::string s);
// implemented by derived classes to clean the task depending on the given
// clean flags; no-op in this class
//
virtual void do_clean(clean);
// implemented by derived classes to fetch (download, clone, etc.) the
// required files to build and/or install the task; no-op in this class
//
virtual void do_fetch();
// implemented by derived classes to build and install the task in one step;
// no-op in this class
//
virtual void do_build_and_install();
// returns the task's context
//
// since a task may be running several threads, a list of per-thread context
// objects is kept in contexts_ and the correct one is returned
//
const context& cx() const;
// throws if the interrupted flag is set
//
void check_interrupted();
virtual void do_clean(clean) {};
virtual void do_fetch() {}
virtual void do_build_and_install() {}
// adds the tool to the internal list of active tools so they can be
// interrupted properly and calls run() on it with the task's log context
//
// once run() returns, removes the tool from the list and returns whatever
// result() returns, which may be void
//
template <class Tool>
auto run_tool(Tool&& t)
{
@@ -70,45 +150,114 @@ protected:
return t.result();
}
void threaded_run(std::string name, std::function<void ()> f);
void parallel(std::vector<std::pair<std::string, std::function<void ()>>> v);
// runs the given functions in a thread_pool with `threads` as the maximum
// number of threads
//
// calls threaded_run() for every function, which creates a new log context
// for the thread in case multiple tools are run simultaneously
//
// this is the preferred way for tasks to run tools in parallel, such as
// in the translations or gtest tasks
//
void parallel(parallel_functions v, std::optional<std::size_t> threads={});
// returns the conf_task for this task, short for conf().task(names())
//
conf_task task_conf() const;
git make_git(git::ops o=git::clone_or_pull) const;
// returns a git tool suitable for this task, with all the relevant task
// settings set, such as ignore_ts_on_clone, shallow, etc.
//
// if `o` is clone_or_pull and the configuration has no_pull for this task,
// it is changed to clone only
//
git make_git() const;
// returns a git url for the given org and repo, using the git_url_prefix
// for this task
//
std::string make_git_url(
const std::string& org, const std::string& repo) const;
private:
// a struct with the thread id and a context object, defined in task.cpp
// to avoid pulling to many includes
//
struct thread_context;
std::vector<std::string> names_;
std::thread thread_;
// names for this task
const std::vector<std::string> names_;
// set when interrupt() is called, checked by check_interrupted(), which
// throws an `interrupted` exception
//
std::atomic<bool> interrupted_;
// holds a context per thread, added/removed in threaded_run()
std::vector<std::unique_ptr<thread_context>> contexts_;
mutable std::mutex contexts_mutex_;
// list of active tools, added/removed in run_tool()
std::vector<tool*> tools_;
mutable std::mutex tools_mutex_;
clean make_clean_flags() const;
// called by run_tool, does the actual work
//
void run_tool_impl(tool* t);
// called by name_matches() when the pattern is a glob
//
bool name_matches_glob(std::string_view pattern) const;
// called by name_matches() when the pattern is not a glob
//
bool name_matches_string(std::string_view pattern) const;
// called by name_matches_string(), compares the two strings to get the
// same result as name_matches_glob() (case insensitive, dashes/underscores
// are the same, etc.) but without requiring a regex because it's slow as
// frick
//
bool strings_match(std::string_view a, std::string_view b) const;
void clean_task();
// called by run() and parallel(), adds a new context for the current thread
// and calls f()
//
void threaded_run(std::string name, std::function<void ()> f);
// calls clean_task(), then do_fetch() if needed (see --no-fetch-task);
// no-op if the task is disabled
//
void fetch();
// calls do_build_and_install() if building is enabled
// (see --no-build-task); no-op if the task is disabled
//
void build_and_install();
// calls do_clean() if needed with the appropriate flags (see
// --no-clean-task); no-op if the task is disabled
//
void clean_task();
};
MOB_ENUM_OPERATORS(task::clean);
// all tasks except for modorganizer have static functions source_path() and
// prebuilt(), which are used in a variety of places, but `task` also needs to
// know about them
//
// so get_source_path() and get_prebuilt() just forward to the static version
// and basic_task uses CRTP so tasks don't have to implement both
//
// since the modorganizer task is reused for all super projects, it doesn't
// have the static member functions and implement these two functions itself
//
template <class Task>
class basic_task : public task
{
@@ -120,37 +269,28 @@ public:
return Task::source_path();
}
std::string get_version() const override
{
return Task::version();
}
const bool get_prebuilt() const override
bool get_prebuilt() const override
{
return Task::prebuilt();
}
};
class container_task : public task
{
public:
using task::task;
virtual std::vector<task*> children() const = 0;
};
class parallel_tasks : public container_task
// a task that overrides run() to start as many threads as it has children
// and calls run() on all of them
//
class parallel_tasks : public task
{
public:
parallel_tasks();
bool enabled() const override
{
return true;
}
// joins
//
~parallel_tasks();
// creates a task `Task`, forwards args to constructor and adds it
//
template <class Task, class... Args>
parallel_tasks& add_task(Args&&... args)
{
@@ -158,6 +298,12 @@ public:
return *this;
}
// creates a task `Task`, forwards args to constructor and adds it
//
// this overload is convenient for modorganizer tasks to pass the task names
// as an initializer list, which can't be done with the version above
// because `Args` can't be deduced
//
template <class Task, class T, class... Args>
parallel_tasks& add_task(std::initializer_list<T> il, Args&&... args)
{
@@ -167,32 +313,38 @@ public:
return *this;
}
// called by the above, adds the task
//
void add_task(std::unique_ptr<task> t);
fs::path get_source_path() const override
{
return {};
}
std::string get_version() const override
{
return {};
}
const bool get_prebuilt() const override
{
return false;
}
// returns true, parallel tasks cannot be disabled, but their children can
//
bool enabled() const override;
// starts a thread for every child task and calls run() on it
//
void run() override;
void interrupt() override;
void join() override;
std::vector<task*> children() const override;
// calls interrupt() on all children tasks
//
void interrupt() override;
// returns children tasks
//
std::vector<task*> children() const;
private:
// tasks
std::vector<std::unique_ptr<task>> children_;
// one thread per task
std::vector<std::thread> threads_;
// joins all threads
//
void join();
};
} // namespace
+1
View File
@@ -1,6 +1,7 @@
#include "pch.h"
#include "task_manager.h"
#include "task.h"
#include "../core/context.h"
namespace mob
{
+1
View File
@@ -1,6 +1,7 @@
#pragma once
#include "task.h"
#include "../tools/tools.h"
#include "../net.h"
#include "../utility.h"
#include "../core/conf.h"
+9 -10
View File
@@ -236,24 +236,23 @@ void translations::do_build_and_install()
for (auto&& w : ps.warnings())
cx().warning(context::generic, "{}", w);
thread_pool tp;
parallel_functions v;
for (auto& p : ps.get())
{
for (auto& lg : p.langs)
{
tp.add([&]
v.push_back({lg.name + "." + p.name, [&]
{
threaded_run(lg.name + "." + p.name, [&]
{
run_tool(lrelease()
.project(p.name)
.sources(lg.ts_files)
.out(dest));
});
});
run_tool(lrelease()
.project(p.name)
.sources(lg.ts_files)
.out(dest));
}});
}
}
parallel(v);
}
} // namespace
+10 -3
View File
@@ -179,8 +179,15 @@ void set_thread_exception_handlers()
}
thread_pool::thread_pool(std::size_t count)
: count_(std::max<std::size_t>(1, count))
std::size_t make_thread_count(std::optional<std::size_t> count)
{
static const auto def = std::thread::hardware_concurrency();
return std::max<std::size_t>(1, count.value_or(def));
}
thread_pool::thread_pool(std::optional<std::size_t> count)
: count_(make_thread_count(count))
{
for (std::size_t i=0; i<count_; ++i)
threads_.emplace_back(std::make_unique<thread_info>());
@@ -226,7 +233,7 @@ bool thread_pool::try_add(fun thread_fun)
t->running = true;
t->thread_fun = thread_fun;
t->thread = std::thread([&]
t->thread = start_thread([&]
{
t->thread_fun();
t->running = false;
+4 -1
View File
@@ -28,7 +28,10 @@ class thread_pool
public:
typedef std::function<void ()> fun;
thread_pool(std::size_t count=std::thread::hardware_concurrency());
thread_pool(std::optional<std::size_t> count={});
// joins
//
~thread_pool();
// non-copyable