thread.h

Go to the documentation of this file.

// Copyright 2016-2026:
//   GobySoft, LLC (2013-)
//   Community contributors (see AUTHORS file)
// File authors:
//   Toby Schneider <toby@gobysoft.org>
//   James D. Turner <james.turner@nrl.navy.mil>
//
//
// This file is part of the Goby Underwater Autonomy Project Libraries
// ("The Goby Libraries").
//
// The Goby Libraries are free software: you can redistribute them and/or modify
// them under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 2.1 of the License, or
// (at your option) any later version.
//
// The Goby Libraries are distributed in the hope that they will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with Goby.  If not, see <http://www.gnu.org/licenses/>.
 
#ifndef GOBY_MIDDLEWARE_APPLICATION_THREAD_H
#define GOBY_MIDDLEWARE_APPLICATION_THREAD_H
 
#include <atomic>
#include <chrono>
#include <memory>
#include <mutex>
#include <typeindex>
 
#include <boost/units/systems/si.hpp>
 
#include "goby/exception.h"
#include "goby/middleware/marshalling/interface.h"
#include "goby/middleware/protobuf/coroner.pb.h"
 
#include "goby/middleware/common.h"
#include "goby/middleware/group.h"
#include "goby/time/simulation.h"
 
namespace goby
{
namespace middleware
{
struct ThreadIdentifier
{
    std::type_index type_i{std::type_index(typeid(void))};
    int index{-1};
    bool all_threads{false};
};
 
template <typename Config, typename TransporterType> class Thread
{
  private:
    TransporterType* transporter_{nullptr};
 
    boost::units::quantity<boost::units::si::frequency> loop_frequency_;
    std::chrono::steady_clock::time_point loop_time_;
    unsigned long long loop_count_{0};
    const Config cfg_;
    int index_;
    std::atomic<bool>* alive_{nullptr};
    std::type_index type_i_{std::type_index(typeid(void))};
 
    // Linux/Apple TID, from gettid()
#ifdef __APPLE__
    uint64_t thread_id_;
#else
    int thread_id_;
#endif
    // demangled class name / index
    std::string thread_name_;
 
    // unique id value used by MultiThreadApplication
    int uid_;
 
    bool finalize_run_{false};
 
  public:
    using Transporter = TransporterType;
 
    Thread(const Config& cfg, TransporterType* transporter, int index)
        : Thread(cfg, transporter, 0 * boost::units::si::hertz, index)
    {
    }
 
    Thread(const Config& cfg, TransporterType* transporter, double loop_freq_hertz = 0,
           int index = -1)
        : Thread(cfg, transporter, loop_freq_hertz * boost::units::si::hertz, index)
    {
    }
 
    Thread(const Config& cfg, TransporterType* transporter,
           boost::units::quantity<boost::units::si::frequency> loop_freq, int index = -1)
        : Thread(cfg, loop_freq, index)
    {
        set_transporter(transporter);
    }
 
    virtual ~Thread() {}
 
    void run(std::atomic<bool>& alive)
    {
        alive_ = &alive;
        do_subscribe();
        initialize();
        while (alive) { run_once(); }
        if (!finalize_run_)
        {
            finalize();
            finalize_run_ = true;
        }
    }
 
    int index() const { return index_; }
 
    std::type_index type_index() { return type_i_; }
    void set_type_index(std::type_index type_i) { type_i_ = type_i; }
 
    std::string name() { return thread_name_; }
    void set_name(const std::string& name) { thread_name_ = name; }
 
    int uid() { return uid_; }
    void set_uid(int uid) { uid_ = uid; }
 
    double loop_frequency_hertz() const { return loop_frequency_ / boost::units::si::hertz; }
    decltype(loop_frequency_) loop_frequency() const { return loop_frequency_; }
 
    void set_loop_frequency_hertz(double loop_freq_hertz)
    {
        set_loop_frequency(loop_freq_hertz * boost::units::si::hertz);
    }
 
    void set_loop_frequency(boost::units::quantity<boost::units::si::frequency> loop_freq)
    {
        loop_frequency_ = loop_freq;
        loop_time_ = std::chrono::steady_clock::now();
 
        if (loop_frequency_hertz() > 0 &&
            loop_frequency_hertz() != std::numeric_limits<double>::infinity())
        {
            unsigned long long microsec_interval =
                1000000.0 / (loop_frequency_hertz() * time::SimulatorSettings::warp_factor);
 
            unsigned long long ticks_since_epoch =
                std::chrono::duration_cast<std::chrono::microseconds>(loop_time_.time_since_epoch())
                    .count() /
                microsec_interval;
 
            loop_time_ = std::chrono::steady_clock::time_point(
                std::chrono::microseconds((ticks_since_epoch + 1) * microsec_interval));
        }
    }
 
    static constexpr goby::middleware::Group shutdown_group_{"goby::middleware::Thread::shutdown"};
    static constexpr goby::middleware::Group joinable_group_{"goby::middleware::Thread::joinable"};
 
  protected:
    Thread(const Config& cfg, boost::units::quantity<boost::units::si::frequency> loop_freq,
           int index = -1)
        : loop_frequency_(loop_freq),
          loop_time_(std::chrono::steady_clock::now()),
          cfg_(cfg),
          index_(index),
          thread_id_(goby::middleware::gettid()),
          thread_name_(std::to_string(thread_id_)),
          uid_(-1)
    {
        set_loop_frequency(loop_freq);
    }
 
    void set_transporter(TransporterType* transporter) { transporter_ = transporter; }
 
    virtual void loop()
    {
        throw(std::runtime_error(
            "void Thread::loop() must be overridden for non-zero loop frequencies"));
    }
 
    double loop_max_frequency() const { return std::numeric_limits<double>::infinity(); }
    void run_once();
 
    TransporterType& transporter() const { return *transporter_; }
 
    const Config& cfg() const { return cfg_; }
 
    // called after alive() is true, but before run()
    virtual void initialize() {}
 
    // called after alive() is false
    virtual void finalize() {}
 
    void thread_health(goby::middleware::protobuf::ThreadHealth& health)
    {
        health.set_name(thread_name_);
#ifdef __APPLE__
        health.set_thread_id_apple(thread_id_);
#else
        health.set_thread_id(thread_id_);
#endif
        if (uid_ >= 0)
            health.set_uid(uid_);
        this->health(health);
    }
 
    virtual void health(goby::middleware::protobuf::ThreadHealth& health)
    {
        health.set_state(goby::middleware::protobuf::HEALTH__OK);
    }
 
    void thread_quit()
    {
        (*alive_) = false;
        if (!finalize_run_)
        {
            finalize();
            finalize_run_ = true;
        }
    }
 
    bool alive() { return alive_ && *alive_; }
 
  private:
    void do_subscribe()
    {
        if (!transporter_)
        {
            throw(goby::Exception(
                "Thread::transporter_ is null. Must set_transporter() before using"));
        }
 
        transporter()
            .innermost()
            .template subscribe<shutdown_group_, ThreadIdentifier, MarshallingScheme::CXX_OBJECT>(
                [this](const ThreadIdentifier ti)
                {
                    if (ti.all_threads ||
                        (ti.type_i == this->type_index() && ti.index == this->index()))
                    {
                        this->thread_quit();
                    }
                });
    }
};
 
} // namespace middleware
 
template <typename Config, typename TransporterType>
constexpr goby::middleware::Group
    goby::middleware::Thread<Config, TransporterType>::shutdown_group_;
 
template <typename Config, typename TransporterType>
constexpr goby::middleware::Group
    goby::middleware::Thread<Config, TransporterType>::joinable_group_;
 
template <typename Config, typename TransporterType>
void goby::middleware::Thread<Config, TransporterType>::run_once()
{
    if (!transporter_)
        throw(goby::Exception("Null transporter"));
 
    if (loop_frequency_hertz() == std::numeric_limits<double>::infinity())
    {
        // call loop as fast as possible
        transporter_->poll(std::chrono::seconds(0));
        loop();
    }
    else if (loop_frequency_hertz() > 0)
    {
        int events = transporter_->poll(loop_time_);
 
        // timeout
        if (events == 0)
        {
            loop();
            ++loop_count_;
            loop_time_ += std::chrono::nanoseconds(
                (unsigned long long)(1000000000ull / (loop_frequency_hertz() *
                                                      time::SimulatorSettings::warp_factor)));
        }
    }
    else
    {
        // don't call loop()
        transporter_->poll();
    }
}
} // namespace goby
 
#endif