driver_thread.h

Go to the documentation of this file.

// Copyright 2019-2026:
//   GobySoft, LLC (2013-)
//   Community contributors (see AUTHORS file)
// File authors:
//   Toby Schneider <toby@gobysoft.org>
//
//
// 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_TRANSPORT_INTERVEHICLE_DRIVER_THREAD_H
#define GOBY_MIDDLEWARE_TRANSPORT_INTERVEHICLE_DRIVER_THREAD_H
 
#include <cstddef>
#include <map>
#include <memory>
#include <ostream>
#include <set>
#include <string>
#include <vector>
 
#include <boost/units/quantity.hpp>
 
#include "goby/acomms/amac/mac_manager.h"
#include "goby/acomms/buffer/dynamic_buffer.h"
#include "goby/acomms/modemdriver/driver_base.h"
#include "goby/acomms/modemdriver/driver_exception.h"
#include "goby/acomms/protobuf/modem_driver_status.pb.h"
#include "goby/middleware/application/thread.h"
#include "goby/middleware/group.h"
#include "goby/middleware/marshalling/dccl.h"
#include "goby/middleware/marshalling/interface.h"
#include "goby/middleware/protobuf/intervehicle.pb.h"
#include "goby/middleware/protobuf/intervehicle_transporter_config.pb.h"
#include "goby/middleware/protobuf/serializer_transporter.pb.h"
#include "goby/middleware/transport/interprocess.h"
#include "goby/middleware/transport/interthread.h"
#include "goby/time/convert.h"
#include "goby/time/steady_clock.h"
#include "goby/time/system_clock.h"
#include "goby/time/types.h"
#include "goby/util/debug_logger/flex_ostream.h"
 
namespace goby
{
namespace acomms
{
namespace protobuf
{
class ModemTransmission;
} // namespace protobuf
} // namespace acomms
 
namespace middleware
{
template <typename Data> class Publisher;
 
namespace protobuf
{
inline size_t data_size(const SerializerTransporterMessage& msg) { return msg.data().size(); }
 
inline bool operator==(const SerializerTransporterMessage& a, const SerializerTransporterMessage& b)
{
    return (a.key().serialize_time() == b.key().serialize_time() &&
            a.key().marshalling_scheme() == b.key().marshalling_scheme() &&
            a.key().type() == b.key().type() && a.key().group() == b.key().group() &&
            a.data() == b.data());
}
 
inline bool operator<(const SerializerTransporterMessage& a, const SerializerTransporterMessage& b)
{
    if (a.key().serialize_time() != b.key().serialize_time())
        return a.key().serialize_time() < b.key().serialize_time();
    else if (a.key().marshalling_scheme() != b.key().marshalling_scheme())
        return a.key().marshalling_scheme() < b.key().marshalling_scheme();
    else if (a.key().type() != b.key().type())
        return a.key().type() < b.key().type();
    else if (a.key().group() != b.key().group())
        return a.key().group() < b.key().group();
    else
        return a.data() < b.data();
}
 
} // namespace protobuf
 
namespace intervehicle
{
namespace protobuf
{
inline bool operator==(const TransporterConfig& a, const TransporterConfig& b)
{
    return a.SerializeAsString() == b.SerializeAsString();
}
 
} // namespace protobuf
 
template <typename Data>
std::shared_ptr<goby::middleware::protobuf::SerializerTransporterMessage>
serialize_publication(const Data& d, const Group& group, const Publisher<Data>& publisher)
{
    std::vector<char> bytes(SerializerParserHelper<Data, MarshallingScheme::DCCL>::serialize(d));
    auto* sbytes = new std::string(bytes.begin(), bytes.end());
    auto msg = std::make_shared<goby::middleware::protobuf::SerializerTransporterMessage>();
 
    auto* key = msg->mutable_key();
    key->set_marshalling_scheme(MarshallingScheme::DCCL);
    key->set_type(SerializerParserHelper<Data, MarshallingScheme::DCCL>::type_name(d));
    key->set_group(std::string(group));
    key->set_group_numeric(group.numeric());
    auto now = goby::time::SystemClock::now<goby::time::MicroTime>();
    key->set_serialize_time_with_units(now);
    *key->mutable_cfg() = publisher.cfg();
    msg->set_allocated_data(sbytes);
    return msg;
}
 
template <typename ImplementationTag>
class ModemDriverThread : public goby::middleware::Thread<
                              intervehicle::protobuf::PortalConfig::LinkConfig,
                              InterProcessForwarder<InterThreadTransporter, ImplementationTag>>
{
  public:
    using buffer_data_type = goby::middleware::protobuf::SerializerTransporterMessage;
    using modem_id_type = goby::acomms::DynamicBuffer<buffer_data_type>::modem_id_type;
    using subbuffer_id_type = goby::acomms::DynamicBuffer<buffer_data_type>::subbuffer_id_type;
    using buffer_value_type = goby::acomms::DynamicBuffer<buffer_data_type>::Value;
 
    ModemDriverThread(const intervehicle::protobuf::PortalConfig::LinkConfig& cfg);
    void loop() override;
    int tx_queue_size() { return buffer_.size(); }
 
  private:
    void finalize() override;
    void _data_request(goby::acomms::protobuf::ModemTransmission* msg);
    void _buffer_message(
        const std::shared_ptr<const goby::middleware::protobuf::SerializerTransporterMessage>& msg);
    void _receive(const goby::acomms::protobuf::ModemTransmission& rx_msg);
    void _forward_subscription(intervehicle::protobuf::Subscription subscription);
    void _accept_subscription(const intervehicle::protobuf::Subscription& subscription);
    void _ack_value(intervehicle::protobuf::AckMessagePair& ack_pair,
                    const goby::time::SteadyClock::time_point now,
                    const goby::acomms::DynamicBuffer<buffer_data_type>::Value& value);
    void _expire_value(const goby::time::SteadyClock::time_point now,
                       const goby::acomms::DynamicBuffer<buffer_data_type>::Value& value,
                       intervehicle::protobuf::ExpireData::ExpireReason reason);
    void _erase_pending_ack(
        const std::vector<goby::acomms::DynamicBuffer<buffer_data_type>::Value>& values);
    void _discard_superseded(
        const std::vector<goby::acomms::DynamicBuffer<buffer_data_type>::Value>& values);
    std::vector<buffer_value_type> _take_superseded(const buffer_value_type& value);
 
    subbuffer_id_type _create_buffer_id(unsigned dccl_id, unsigned group);
 
    subbuffer_id_type
    _create_buffer_id(const goby::middleware::protobuf::SerializerTransporterKey& key)
    {
        return _create_buffer_id(detail::DCCLSerializerParserHelperBase::id(key.type()),
                                 key.group_numeric());
    }
 
    subbuffer_id_type _create_buffer_id(const intervehicle::protobuf::Subscription& subscription)
    {
        return _create_buffer_id(subscription.dccl_id(), subscription.group());
    }
 
    void _try_create_or_update_buffer(modem_id_type dest_id, const subbuffer_id_type& buffer_id);
 
    modem_id_type _broadcast_id() { return this->cfg().modem_id() & this->cfg().subnet_mask(); }
 
    // id within subnet
    modem_id_type _id_within_subnet(modem_id_type id) { return id - _broadcast_id(); }
 
    // full id
    modem_id_type _full_id(modem_id_type id_in_subnet) { return id_in_subnet + _broadcast_id(); }
 
    bool _dest_is_in_subnet(modem_id_type dest_id)
    {
        bool dest_in_subnet = (dest_id & this->cfg().subnet_mask()) ==
                              (this->cfg().modem_id() & this->cfg().subnet_mask());
        if (!dest_in_subnet)
            goby::glog.is_debug3() &&
                goby::glog << "Dest: " << dest_id
                           << " is not in subnet (our id: " << this->cfg().modem_id()
                           << ", mask: " << this->cfg().subnet_mask() << ")" << std::endl;
 
        return dest_in_subnet;
    }
 
    void _publish_subscription_report(const intervehicle::protobuf::Subscription& changed);
    void _create_driver();
    bool _try_driver_startup();
    // returns false if the failure should instead terminate the portal
    bool _restart_on_failure(const goby::acomms::ModemDriverException& e);
    void _publish_modem_report();
 
  private:
    std::unique_ptr<InterThreadTransporter> interthread_;
    std::unique_ptr<InterProcessForwarder<InterThreadTransporter, ImplementationTag>> interprocess_;
 
    std::multimap<subbuffer_id_type, goby::middleware::protobuf::SerializerTransporterKey>
        publisher_buffer_cfg_;
 
    std::map<modem_id_type, std::multimap<subbuffer_id_type, intervehicle::protobuf::Subscription>>
        subscriber_buffer_cfg_;
 
    std::map<subbuffer_id_type, std::set<modem_id_type>> subbuffers_created_;
 
    goby::middleware::protobuf::SerializerTransporterKey subscription_key_;
    std::set<modem_id_type> subscription_subbuffers_;
    struct QueuedSubscription
    {
        intervehicle::protobuf::Subscription subscription;
        buffer_data_type data;
        // equivalent subscriptions (possibly from other local subscribers) that this one replaced
        // and now stands in for: they are acked or expired along with it
        std::vector<buffer_value_type> superseded;
    };
    // subscriptions queued or awaiting ack per destination, so a newer equivalent one can replace them
    std::map<modem_id_type, std::vector<QueuedSubscription>> queued_subscriptions_;
 
    goby::acomms::DynamicBuffer<buffer_data_type> buffer_;
 
    using frame_type = int;
    std::map<frame_type, std::vector<goby::acomms::DynamicBuffer<buffer_data_type>::Value>>
        pending_ack_;
 
    std::unique_ptr<goby::acomms::ModemDriverBase> driver_;
    goby::acomms::MACManager mac_;
 
    std::string glog_group_;
 
    goby::time::SteadyClock::time_point next_modem_report_time_;
    const goby::time::SteadyClock::duration modem_report_interval_;
 
    bool driver_started_{false};
    goby::acomms::protobuf::ModemDriverStatus::Status failure_status_{
        goby::acomms::protobuf::ModemDriverStatus::NOMINAL};
    goby::time::SteadyClock::time_point next_restart_attempt_;
};
 
} // namespace intervehicle
} // namespace middleware
} // namespace goby
 
#endif