actsim-cluster-agent/src/actsim_agent/task_interface.cpp

/*************************************************************************
 *
 *  This file is part of the ACT library
 *
 *  Copyright (c) 2024 Fabian Posch
 *
 *  This program is free software; you can redistribute it and/or
 *  modify it under the terms of the GNU General Public License
 *  as published by the Free Software Foundation; either version 2
 *  of the License, or (at your option) any later version.
 *
 *  This program is distributed in the hope that it will be useful,
 *  but WITHOUT ANY WARRANTY; without even the implied warranty of
 *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *  GNU General Public License for more details.
 *
 *  You should have received a copy of the GNU General Public License
 *  along with this program; if not, write to the Free Software
 *  Foundation, Inc., 51 Franklin Street, Fifth Floor,
 *  Boston, MA  02110-1301, USA.
 *
 **************************************************************************
 */

#include "task_interface.hpp"

TaskInterface::TaskInterface(size_t buffer_size) {
    this->buffer_size = buffer_size;
}

void TaskInterface::push_fresh(std::unique_ptr<InputType> task) {

    // lock the queue and insert into it
    std::lock_guard<std::mutex> lock(this->fresh_queue_mutex);
    this->fresh_queue.push(std::move(task));

    // we put one task in there so we notify one worker thread
    this->fresh_queue_empty_condition.notify_one();
}

bool TaskInterface::fresh_queue_empty() {
    std::lock_guard<std::mutex> lock(this->fresh_queue_mutex);
    return this->fresh_queue.empty();
}

bool TaskInterface::finished_queue_empty() {
    std::lock_guard<std::mutex> lock(this->finished_queue_mutex);
    return this->finished_queue.empty();
}

std::unique_ptr<InputType> TaskInterface::pop_fresh(bool& empty) {

    // we first need exclusive access to the queue
    std::lock_guard<std::mutex> lock (this->fresh_queue_mutex);

    // we have to make sure the queue wasn't emptied since the last empty call was made
    // or at least inform the calling thread that the queue is currently empty
    std::unique_ptr<InputType> task;

    if (!(empty = this->fresh_queue.empty())) {
        task = std::move(this->fresh_queue.front());
        
        // if the task needs to be simulated by multiple workers, don't remove it from the queue
        if (!task->parallelizable()) {
            this->fresh_queue.pop();
            this->fresh_queue_full_condition.notify_one();
        }
    }

    return task;
}

void TaskInterface::wait_for_fresh() {
    std::unique_lock<std::mutex> lock (this->fresh_queue_empty_mutex);

    // we will be notified either when there is new data or the program has been stopped
    this->fresh_queue_empty_condition.wait(lock, [&] { return !this->fresh_queue_empty() || !running(); });
}

void TaskInterface::wait_for_finished() {
    std::unique_lock<std::mutex> lock (this->finished_queue_empty_mutex);

    // we will be notified either when there is new data or the program has been stopped
    this->finished_queue_empty_condition.wait(lock, [&] { return !this->finished_queue_empty() || !running(); });
}

void TaskInterface::wait_for_buffer_consume() {
    std::unique_lock<std::mutex> lock (this->fresh_queue_full_mutex);

    // we will be notified either when there is new data or the program has been stopped
    this->fresh_queue_full_condition.wait(lock, [&] { return this->get_buffer_space() > 0 || !running(); });
}

void TaskInterface::push_finished(std::unique_ptr<OutputType> task) {
    std::lock_guard<std::mutex> lock(this->finished_queue_mutex);
    this->finished_queue.push(std::move(task));

    // there is new data, inform the upload thread
    this->finished_queue_empty_condition.notify_one();
}

std::unique_ptr<OutputType> TaskInterface::pop_finished(bool& empty) {

    // we first need exclusive access to the queue
    std::lock_guard<std::mutex> lock (this->finished_queue_mutex);

    // we have to make sure the queue wasn't emptied since the last empty call was made
    // or at least inform the calling thread that the queue is currently empty
    std::unique_ptr<OutputType> task;

    if (!(empty = this->finished_queue.empty())) {
        auto task = std::move(this->finished_queue.front());
        this->finished_queue.pop();
    }

    return task;
}

bool TaskInterface::search_and_increment(db::uuid_t id, std::string& design) {
    std::lock_guard<std::mutex> lock (this->designs_mutex);

    // make sure the requested design is in the list of available designs
    if (this->designs.find(id) == this->designs.end()) return false;

    std::pair<size_t, std::string>& design_entry = designs[id];

    // if so, increment its reference counters
    ++design_entry.first;

    // set the design parameter to the actual design information
    design = design_entry.second;
    return true;
}

void TaskInterface::decrement(db::uuid_t id) {
    std::lock_guard<std::mutex> lock (this->designs_mutex);

    // make sure the requested design is in the list of available designs
    if (this->designs.find(id) == this->designs.end()) return;

    std::pair<size_t, std::string>& design_entry = designs[id];

    // if so, decrement its reference counters
    --design_entry.first;

    // if the reference counter hit 0, erase the design entry from the list
    // of available designs
    if (design_entry.first == 0) {
        this->designs.erase(id);
    }
}

void TaskInterface::store(db::uuid_t id, std::string& design) {
    std::lock_guard<std::mutex> lock (this->designs_mutex);

    // make sure the design isn't already in the list of design entries
    // if it is, just increment its reference counter
    if (this->designs.find(id) != this->designs.end()) {
        ++(this->designs[id]).first;
        return;
    }

    // otherwise, create a new entry for this design
    this->designs[id] = {1, design};
}

size_t TaskInterface::get_buffer_space() {
    std::lock_guard<std::mutex> lock(this->fresh_queue_mutex);
    return this->buffer_size - this->fresh_queue.size();
}

void TaskInterface::notify_cleanup_ready() {
    this->cleanup_ready.store(true, std::memory_order_seq_cst);
    this->cleanup_ready_condition.notify_all();
}

void TaskInterface::wait_for_cleanup_ready() {
    // lock the thread and wait to be notified
    std::unique_lock<std::mutex> lock(this->cleanup_ready_mutex);
    this->cleanup_ready_condition.wait(lock, [&] { return this->cleanup_ready.load(std::memory_order_seq_cst); });
}

void TaskInterface::notify_download_program_halt() {
    this->fresh_queue_full_condition.notify_all();
}

void TaskInterface::notify_workers_program_halt() {
    this->finished_queue_empty_condition.notify_all();
    this->fresh_queue_empty_condition.notify_all();
}