fix join null check and expose current task uuid in worker

This commit is contained in:
Fabian Posch 2024-01-10 15:45:31 -05:00
parent 0c8eb1c21d
commit 2730d204f9
2 changed files with 15 additions and 5 deletions

View file

@ -37,7 +37,7 @@ void Uploader::start() {
} }
void Uploader::join() { void Uploader::join() {
if (this->uploader_thread != nullptr) return; if (this->uploader_thread == nullptr) return;
this->uploader_thread->join(); this->uploader_thread->join();
} }
@ -99,6 +99,9 @@ void Uploader::thread_run() {
} }
bool Uploader::upload_task([[maybe_unused]]std::unique_ptr<Task> task) { bool Uploader::upload_task([[maybe_unused]]std::unique_ptr<Task> task) {
// make sure any task that is uploaded isn't halted in the database
std::cout << "Task uploaded!"; std::cout << "Task uploaded!";
return true; return true;
} }

View file

@ -31,15 +31,17 @@ Worker::Worker(TaskInterface& interface) : interface(interface) {}
void Worker::start() { void Worker::start() {
std::cout << "Worker started" << std::endl; std::cout << "Worker started" << std::endl;
this->worker_thread = std::make_unique<std::thread>([this]() { thread_run(); }); this->worker_thread = std::make_unique<std::thread>([this] () { this->thread_run(); });
} }
void Worker::cancel_current() { void Worker::cancel_current() {
std::cout << "Current simulation cancelled." << std::endl; std::cout << "Current simulation cancelled." << std::endl;
// set a condition variable and notify
// must not be blocking
} }
void Worker::join() { void Worker::join() {
if (this->worker_thread != nullptr) return; if (this->worker_thread == nullptr) return;
this->worker_thread->join(); this->worker_thread->join();
} }
@ -55,12 +57,14 @@ void Worker::thread_run() {
// we're still good to go! get a task from the fresh queue // we're still good to go! get a task from the fresh queue
bool queue_empty; bool queue_empty;
auto task = this->interface.pop_fresh(queue_empty); auto task = this->interface.pop_fresh(queue_empty);
this->current_task.store(task->uuid, std::memory_order_relaxed);
// we need to make sure the queue wasn't emptied between waiting and getting new data // we need to make sure the queue wasn't emptied between waiting and getting new data
if (queue_empty) continue; if (queue_empty) continue;
// everything is good, perform the given task // everything is good, perform the given task
bool complete = this->perform_task(task); bool complete = this->perform_task(task);
this->current_task.store(db::uuid_t(), std::memory_order_relaxed);
// if the task was finished, push the task to be uploaded // if the task was finished, push the task to be uploaded
// we need this since the task might have been interrupted // we need this since the task might have been interrupted
@ -70,8 +74,11 @@ void Worker::thread_run() {
} }
bool Worker::perform_task(std::unique_ptr<Task>& task) { bool Worker::perform_task(std::unique_ptr<Task>& task) {
std::cout << "Worker ping" << std::endl;
usleep(100000); usleep(100000);
task->get_uuid(); std::cout << "Worker performed task. Please implement me!" << std::endl;
task->uuid;
return true; return true;
// wait for either the executing process to finish or
// for the condition variable indicating the process should be stopped
} }