IOThread.hpp

Go to the documentation of this file.

// SPDX-License-Identifier: Apache-2.0
//
// Copyright 2026 Caleb Buahin
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

 
#ifndef OPENSWMM_ENGINE_IO_THREAD_HPP
#define OPENSWMM_ENGINE_IO_THREAD_HPP
 
#include "WriteTask.hpp"
 
#include <thread>
#include <mutex>
#include <condition_variable>
#include <queue>
#include <atomic>
#include <functional>

 
namespace openswmm {
 
class PluginFactory;

class IOThread {
public:
    static constexpr std::size_t DEFAULT_QUEUE_CAPACITY = 8;

    explicit IOThread(PluginFactory& factory, std::size_t capacity = DEFAULT_QUEUE_CAPACITY);

    ~IOThread();
 
    // Non-copyable, non-movable (owns a thread)
    IOThread(const IOThread&) = delete;
    IOThread& operator=(const IOThread&) = delete;
 
    // -----------------------------------------------------------------------
    // Control
    // -----------------------------------------------------------------------

    void start();

    void post(SimulationSnapshot snap);

    void stop();

    int flush() const;
 
    // -----------------------------------------------------------------------
    // Diagnostics
    // -----------------------------------------------------------------------

    bool running() const noexcept { return running_.load(std::memory_order_relaxed); }

    int tasks_completed() const noexcept {
        return tasks_completed_.load(std::memory_order_relaxed);
    }

    int last_error() const noexcept { return last_error_.load(std::memory_order_relaxed); }
 
private:
    void run(); 
 
    PluginFactory& factory_;
    const std::size_t capacity_;
 
    std::thread thread_;
    std::queue<WriteTask> queue_;
    mutable std::mutex mutex_;
    mutable std::condition_variable cv_idle_;
    bool writing_ = false;
    std::condition_variable cv_not_full_;  
    std::condition_variable cv_not_empty_; 
    std::atomic<bool> stop_flag_{false};
    std::atomic<bool> running_{false};
    std::atomic<int> tasks_completed_{0};
    std::atomic<int> last_error_{0};
    int next_sequence_ = 0;
};
 
} /* namespace openswmm */
 
#endif /* OPENSWMM_ENGINE_IO_THREAD_HPP */