1
0
mirror of https://github.com/Zygo/bees.git synced 2025-05-18 05:45:45 +02:00

Task: convert print_fn to a string

Since we are now unconditionally rendering the print_fn as a static
string, there is no need for it to be a function.  We also need it to
be brief and mostly constant.

Use a string instead.  Put the string before the function in the Task
constructor arguments so that the title string appears as a heading in
code, since we are making a breaking API change already.

Drop TASK_MACRO as it is broken by this change, but there is no similar
usage of Task anywhere to make it worth fixing.

Signed-off-by: Zygo Blaxell <bees@furryterror.org>
This commit is contained in:
Zygo Blaxell 2018-01-25 21:12:31 -05:00
parent 0710208354
commit f64fc78e36
5 changed files with 75 additions and 86 deletions

View File

@ -4,6 +4,7 @@
#include <functional> #include <functional>
#include <memory> #include <memory>
#include <ostream> #include <ostream>
#include <string>
namespace crucible { namespace crucible {
using namespace std; using namespace std;
@ -22,7 +23,7 @@ namespace crucible {
public: public:
// create Task object containing closure and description // create Task object containing closure and description
Task(function<void()> exec_fn, function<ostream&(ostream &)> print_fn); Task(string title, function<void()> exec_fn);
// schedule Task at end of queue. // schedule Task at end of queue.
// May run Task in current thread or in other thread. // May run Task in current thread or in other thread.
@ -33,7 +34,7 @@ namespace crucible {
void run_earlier() const; void run_earlier() const;
// describe Task as text // describe Task as text
ostream &print(ostream &os) const; string title() const;
// Returns currently executing task if called from exec_fn. // Returns currently executing task if called from exec_fn.
// Usually used to reschedule the currently executing Task. // Usually used to reschedule the currently executing Task.

View File

@ -19,15 +19,15 @@ namespace crucible {
class TaskState : public enable_shared_from_this<TaskState> { class TaskState : public enable_shared_from_this<TaskState> {
const function<void()> m_exec_fn; const function<void()> m_exec_fn;
const function<ostream&(ostream &)> m_print_fn; const string m_title;
TaskId m_id; TaskId m_id;
static atomic<TaskId> s_next_id; static atomic<TaskId> s_next_id;
public: public:
TaskState(function<void()> exec_fn, function<ostream&(ostream &)> print_fn); TaskState(string title, function<void()> exec_fn);
void exec(); void exec();
ostream &print(ostream &os); string title() const;
TaskId id() const; TaskId id() const;
}; };
@ -73,19 +73,19 @@ namespace crucible {
static shared_ptr<TaskMasterState> s_tms = make_shared<TaskMasterState>(); static shared_ptr<TaskMasterState> s_tms = make_shared<TaskMasterState>();
TaskState::TaskState(function<void()> exec_fn, TaskState::TaskState(string title, function<void()> exec_fn) :
function<ostream&(ostream &)> print_fn) :
m_exec_fn(exec_fn), m_exec_fn(exec_fn),
m_print_fn(print_fn), m_title(title),
m_id(++s_next_id) m_id(++s_next_id)
{ {
THROW_CHECK0(invalid_argument, !m_title.empty());
} }
void void
TaskState::exec() TaskState::exec()
{ {
THROW_CHECK0(invalid_argument, m_exec_fn); THROW_CHECK0(invalid_argument, m_exec_fn);
THROW_CHECK0(invalid_argument, m_print_fn); THROW_CHECK0(invalid_argument, !m_title.empty());
char buf[24]; char buf[24];
memset(buf, '\0', sizeof(buf)); memset(buf, '\0', sizeof(buf));
@ -93,10 +93,7 @@ namespace crucible {
Cleanup pthread_name_cleaner([&]() { Cleanup pthread_name_cleaner([&]() {
pthread_setname_np(pthread_self(), buf); pthread_setname_np(pthread_self(), buf);
}); });
ostringstream oss; DIE_IF_MINUS_ERRNO(pthread_setname_np(pthread_self(), m_title.c_str()));
m_print_fn(oss);
auto thread_name = oss.str();
DIE_IF_MINUS_ERRNO(pthread_setname_np(pthread_self(), thread_name.c_str()));
weak_ptr<TaskState> this_task_wp = shared_from_this(); weak_ptr<TaskState> this_task_wp = shared_from_this();
Cleanup current_task_cleaner([&]() { Cleanup current_task_cleaner([&]() {
@ -107,11 +104,11 @@ namespace crucible {
m_exec_fn(); m_exec_fn();
} }
ostream & string
TaskState::print(ostream &os) TaskState::title() const
{ {
THROW_CHECK0(invalid_argument, m_print_fn); THROW_CHECK0(runtime_error, !m_title.empty());
return m_print_fn(os); return m_title;
} }
TaskId TaskId
@ -176,9 +173,7 @@ namespace crucible {
os << "Queue (size " << s_tms->m_queue.size() << "):" << endl; os << "Queue (size " << s_tms->m_queue.size() << "):" << endl;
size_t counter = 0; size_t counter = 0;
for (auto i : s_tms->m_queue) { for (auto i : s_tms->m_queue) {
os << "Queue #" << ++counter << " Task ID " << i->id() << " "; os << "Queue #" << ++counter << " Task ID " << i->id() << " " << i->title() << endl;
i->print(os);
os << endl;
} }
return os << "Queue End" << endl; return os << "Queue End" << endl;
} }
@ -193,8 +188,7 @@ namespace crucible {
os << "Worker #" << ++counter << " "; os << "Worker #" << ++counter << " ";
auto task = i->current_task_locked(); auto task = i->current_task_locked();
if (task) { if (task) {
os << "Task ID " << task->id() << " "; os << "Task ID " << task->id() << " " << task->title();
task->print(os);
} else { } else {
os << "(idle)"; os << "(idle)";
} }
@ -238,8 +232,8 @@ namespace crucible {
{ {
} }
Task::Task(function<void()> exec_fn, function<ostream&(ostream &)> print_fn) : Task::Task(string title, function<void()> exec_fn) :
m_task_state(make_shared<TaskState>(exec_fn, print_fn)) m_task_state(make_shared<TaskState>(title, exec_fn))
{ {
} }
@ -263,17 +257,17 @@ namespace crucible {
return Task(tl_current_task_wp.lock()); return Task(tl_current_task_wp.lock());
} }
ostream & string
Task::print(ostream &os) const Task::title() const
{ {
THROW_CHECK0(runtime_error, m_task_state); THROW_CHECK0(runtime_error, m_task_state);
return m_task_state->print(os); return m_task_state->title();
} }
ostream & ostream &
operator<<(ostream &os, const Task &task) operator<<(ostream &os, const Task &task)
{ {
return task.print(os); return os << task.title();
}; };
TaskId TaskId

View File

@ -251,12 +251,12 @@ BeesRoots::crawl_roots()
size_t batch_count = 0; size_t batch_count = 0;
while (first_range && batch_count < BEES_MAX_CRAWL_BATCH) { while (first_range && batch_count < BEES_MAX_CRAWL_BATCH) {
auto subvol = first_crawl->get_state().m_root; auto subvol = first_crawl->get_state().m_root;
Task([ctx_copy, first_range]() { ostringstream oss;
oss << "crawl_" << subvol;
auto task_title = oss.str();
Task(task_title, [ctx_copy, first_range]() {
BEESNOTE("scan_forward " << first_range); BEESNOTE("scan_forward " << first_range);
ctx_copy->scan_forward(first_range); ctx_copy->scan_forward(first_range);
},
[first_range, subvol](ostream &os) -> ostream & {
return os << "crawl_" << subvol;
}).run(); }).run();
BEESCOUNT(crawl_scan); BEESCOUNT(crawl_scan);
m_crawl_current = first_crawl->get_state(); m_crawl_current = first_crawl->get_state();
@ -281,12 +281,12 @@ BeesRoots::crawl_roots()
size_t batch_count = 0; size_t batch_count = 0;
while (this_range && batch_count < BEES_MAX_CRAWL_BATCH) { while (this_range && batch_count < BEES_MAX_CRAWL_BATCH) {
auto subvol = this_crawl->get_state().m_root; auto subvol = this_crawl->get_state().m_root;
Task([ctx_copy, this_range]() { ostringstream oss;
oss << "crawl_" << subvol;
auto task_title = oss.str();
Task(task_title, [ctx_copy, this_range]() {
BEESNOTE("scan_forward " << this_range); BEESNOTE("scan_forward " << this_range);
ctx_copy->scan_forward(this_range); ctx_copy->scan_forward(this_range);
},
[this_range, subvol](ostream &os) -> ostream & {
return os << "crawl_" << subvol;
}).run(); }).run();
crawled = true; crawled = true;
BEESCOUNT(crawl_scan); BEESCOUNT(crawl_scan);
@ -322,7 +322,7 @@ BeesRoots::crawl_thread()
// shared_from_this() in a constructor. // shared_from_this() in a constructor.
BEESNOTE("crawling"); BEESNOTE("crawling");
auto shared_this = shared_from_this(); auto shared_this = shared_from_this();
Task([shared_this]() { Task("crawl", [shared_this]() {
auto tqs = TaskMaster::get_queue_count(); auto tqs = TaskMaster::get_queue_count();
BEESNOTE("queueing extents to scan, " << tqs << " of " << BEES_MAX_QUEUE_SIZE); BEESNOTE("queueing extents to scan, " << tqs << " of " << BEES_MAX_QUEUE_SIZE);
while (tqs < BEES_MAX_QUEUE_SIZE) { while (tqs < BEES_MAX_QUEUE_SIZE) {
@ -332,7 +332,7 @@ BeesRoots::crawl_thread()
tqs = TaskMaster::get_queue_count(); tqs = TaskMaster::get_queue_count();
} }
Task::current_task().run(); Task::current_task().run();
}, [](ostream &os) -> ostream& { return os << "crawl"; }).run(); }).run();
} }
void void

View File

@ -151,9 +151,7 @@ BeesNote::get_name()
// about it being destroyed under us. // about it being destroyed under us.
auto current_task = Task::current_task(); auto current_task = Task::current_task();
if (current_task) { if (current_task) {
ostringstream oss; return current_task.title();
oss << current_task;
return oss.str();
} }
// OK try the pthread name next. // OK try the pthread name next.

View File

@ -28,12 +28,17 @@ test_tasks(size_t count)
// Run several tasks in parallel // Run several tasks in parallel
for (size_t c = 0; c < count; ++c) { for (size_t c = 0; c < count; ++c) {
Task t([c, &task_done, &mtx, &cv]() { ostringstream oss;
oss << "task #" << c;
Task t(
oss.str(),
[c, &task_done, &mtx, &cv]() {
unique_lock<mutex> lock(mtx); unique_lock<mutex> lock(mtx);
// cerr << "Task #" << c << endl; // cerr << "Task #" << c << endl;
task_done.at(c) = true; task_done.at(c) = true;
cv.notify_one(); cv.notify_one();
}, [=](ostream& os) -> ostream& { return os << "task #" << c; }); }
);
t.run(); t.run();
} }
@ -58,20 +63,6 @@ test_tasks(size_t count)
} }
} }
#define TASK_MACRO(printer, body) Task( \
[=]() { body; }, \
[=](ostream &__os) -> ostream& { return __os << printer; } \
)
void
test_macro()
{
TASK_MACRO(
"A Task",
cerr << "Hello, World!" << endl;
).run();
}
void void
test_finish() test_finish()
{ {
@ -104,12 +95,17 @@ test_barrier(size_t count)
// Run several tasks in parallel // Run several tasks in parallel
for (size_t c = 0; c < count; ++c) { for (size_t c = 0; c < count; ++c) {
auto bl = b->lock(); auto bl = b->lock();
Task t([c, &task_done, &mtx, &cv, bl]() mutable { ostringstream oss;
oss << "task #" << c;
Task t(
oss.str(),
[c, &task_done, &mtx, &cv, bl]() mutable {
// cerr << "Task #" << c << endl; // cerr << "Task #" << c << endl;
unique_lock<mutex> lock(mtx); unique_lock<mutex> lock(mtx);
task_done.at(c) = true; task_done.at(c) = true;
bl.release(); bl.release();
}, [=](ostream& os) -> ostream& { return os << "task #" << c; }); }
);
t.run(); t.run();
} }
@ -121,14 +117,12 @@ test_barrier(size_t count)
bool done_flag = false; bool done_flag = false;
Task completed( Task completed(
"Waiting for Barrier",
[&mtx, &cv, &done_flag]() { [&mtx, &cv, &done_flag]() {
unique_lock<mutex> lock(mtx); unique_lock<mutex> lock(mtx);
// cerr << "Running cv notify" << endl; // cerr << "Running cv notify" << endl;
done_flag = true; done_flag = true;
cv.notify_all(); cv.notify_all();
},
[](ostream &os) -> ostream& {
return os << "Waiting for Barrier";
} }
); );
b->insert_task(completed); b->insert_task(completed);
@ -168,7 +162,11 @@ test_exclusion(size_t count)
// Run several tasks in parallel // Run several tasks in parallel
for (size_t c = 0; c < count; ++c) { for (size_t c = 0; c < count; ++c) {
auto bl = b->lock(); auto bl = b->lock();
Task t([c, &only_one, &mtx, &excl, bl]() mutable { ostringstream oss;
oss << "task #" << c;
Task t(
oss.str(),
[c, &only_one, &mtx, &excl, bl]() mutable {
// cerr << "Task #" << c << endl; // cerr << "Task #" << c << endl;
auto lock = excl.try_lock(); auto lock = excl.try_lock();
if (!lock) { if (!lock) {
@ -180,21 +178,20 @@ test_exclusion(size_t count)
nanosleep(0.0001); nanosleep(0.0001);
only_one.unlock(); only_one.unlock();
bl.release(); bl.release();
}, [=](ostream& os) -> ostream& { return os << "task #" << c; }); }
);
t.run(); t.run();
} }
bool done_flag = false; bool done_flag = false;
Task completed( Task completed(
"Waiting for Barrier",
[&mtx, &cv, &done_flag]() { [&mtx, &cv, &done_flag]() {
unique_lock<mutex> lock(mtx); unique_lock<mutex> lock(mtx);
// cerr << "Running cv notify" << endl; // cerr << "Running cv notify" << endl;
done_flag = true; done_flag = true;
cv.notify_all(); cv.notify_all();
},
[](ostream &os) -> ostream& {
return os << "Waiting for Barrier";
} }
); );
b->insert_task(completed); b->insert_task(completed);
@ -217,7 +214,6 @@ main(int, char**)
alarm(9); alarm(9);
RUN_A_TEST(test_tasks(256)); RUN_A_TEST(test_tasks(256));
RUN_A_TEST(test_macro());
RUN_A_TEST(test_finish()); RUN_A_TEST(test_finish());
RUN_A_TEST(test_unfinish()); RUN_A_TEST(test_unfinish());
RUN_A_TEST(test_barrier(256)); RUN_A_TEST(test_barrier(256));