dynamic_buffer.h

Go to the documentation of this file.

// Copyright 2019-2023:
//   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_ACOMMS_BUFFER_DYNAMIC_BUFFER_H
#define GOBY_ACOMMS_BUFFER_DYNAMIC_BUFFER_H
 
#include <deque>
#include <type_traits>
#include <unordered_map>
#include <vector>
 
#include "goby/acomms/acomms_constants.h"
#include "goby/acomms/protobuf/buffer.pb.h"
#include "goby/exception.h"
#include "goby/time/convert.h"
#include "goby/time/steady_clock.h"
#include "goby/util/debug_logger.h"
 
namespace goby
{
namespace acomms
{
namespace protobuf
{
inline bool operator==(const DynamicBufferConfig& a, const DynamicBufferConfig& b)
{
    return a.SerializeAsString() == b.SerializeAsString();
}
 
} // namespace protobuf
 
class DynamicBufferNoDataException : goby::Exception
{
  public:
    DynamicBufferNoDataException() : goby::Exception("No queues have data available") {}
    ~DynamicBufferNoDataException() {}
};
 
template <typename Container> size_t data_size(const Container& c) { return c.size(); }
 
template <typename T, typename Clock = goby::time::SteadyClock> class DynamicSubBuffer
{
  public:
    using size_type = typename std::deque<T>::size_type;
 
    struct Value
    {
        typename Clock::time_point push_time;
        T data;
        bool operator==(const Value& a) const { return a.push_time == push_time && a.data == data; }
    };
 
    DynamicSubBuffer(const goby::acomms::protobuf::DynamicBufferConfig& cfg)
        : DynamicSubBuffer(std::vector<goby::acomms::protobuf::DynamicBufferConfig>(1, cfg))
    {
    }
 
    DynamicSubBuffer(const std::vector<goby::acomms::protobuf::DynamicBufferConfig>& cfgs)
    {
        update(cfgs);
    }
 
    ~DynamicSubBuffer() {}
 
    void update(const std::vector<goby::acomms::protobuf::DynamicBufferConfig>& cfgs)
    {
        using goby::acomms::protobuf::DynamicBufferConfig;
        if (cfgs.empty())
            throw(goby::Exception("Configuration vector must not be empty for DynamicSubBuffer"));
 
        cfg_.Clear();
 
        // extract these types from the Protobuf message
        using ttl_type =
            std::result_of<decltype (&DynamicBufferConfig::ttl)(DynamicBufferConfig)>::type;
        using value_base_type =
            std::result_of<decltype (&DynamicBufferConfig::value_base)(DynamicBufferConfig)>::type;
 
        ttl_type ttl_sum = 0;
        ttl_type ttl_divisor = 0;
        value_base_type value_base_sum = 0;
        value_base_type value_base_divisor = 0;
 
        for (const auto& cfg : cfgs)
        {
            if (cfg.has_ack_required() && (!cfg_.has_ack_required() || cfg.ack_required()))
                cfg_.set_ack_required(cfg.ack_required());
            if (cfg.has_blackout_time() &&
                (!cfg_.has_blackout_time() || cfg.blackout_time() < cfg_.blackout_time()))
                cfg_.set_blackout_time(cfg.blackout_time());
            if (cfg.has_max_queue() &&
                (!cfg_.has_max_queue() || cfg.max_queue() > cfg_.max_queue()))
                cfg_.set_max_queue(cfg.max_queue());
            if (cfg.has_newest_first() && (!cfg_.has_newest_first() || cfg.newest_first()))
                cfg_.set_newest_first(cfg.newest_first());
 
            if (cfg.has_ttl())
            {
                ttl_sum += cfg.ttl();
                ++ttl_divisor;
            }
 
            if (cfg.has_value_base())
            {
                value_base_sum += cfg.value_base();
                ++value_base_divisor;
            }
        }
 
        if (ttl_divisor > 0)
            cfg_.set_ttl(ttl_sum / ttl_divisor);
        if (value_base_divisor > 0)
            cfg_.set_value_base(value_base_sum / value_base_divisor);
    }
 
    const goby::acomms::protobuf::DynamicBufferConfig& cfg() const { return cfg_; }
 
    Value& top(typename Clock::time_point reference = Clock::now(),
               typename Clock::duration ack_timeout = std::chrono::microseconds(0))
    {
        for (auto& datum_pair : data_)
        {
            auto& datum_last_access = datum_pair.first;
            if (datum_last_access == zero_point_ || datum_last_access + ack_timeout < reference)
            {
                last_access_ = reference;
                datum_last_access = last_access_;
                return datum_pair.second;
            }
        }
        throw(DynamicBufferNoDataException());
    }
 
    size_t top_size(typename Clock::time_point reference = Clock::now(),
                    typename Clock::duration ack_timeout = std::chrono::microseconds(0)) const
    {
        for (const auto& datum_pair : data_)
        {
            const auto& datum_last_access = datum_pair.first;
            if (datum_last_access == zero_point_ || datum_last_access + ack_timeout < reference)
                return data_size(datum_pair.second.data);
        }
        throw(DynamicBufferNoDataException());
    }
 
    bool
    all_waiting_for_ack(typename Clock::time_point reference = Clock::now(),
                        typename Clock::duration ack_timeout = std::chrono::microseconds(0)) const
    {
        for (const auto& datum_pair : data_)
        {
            const auto& datum_last_access = datum_pair.first;
            if (datum_last_access == zero_point_ || datum_last_access + ack_timeout < reference)
                return false;
        }
        return true;
    }
 
    enum class ValueResult
    {
        VALUE_PROVIDED,
        EMPTY,
        IN_BLACKOUT,
        NEXT_MESSAGE_TOO_LARGE,
        ALL_MESSAGES_WAITING_FOR_ACK
    };
 
    std::pair<double, ValueResult>
    top_value(typename Clock::time_point reference = Clock::now(),
              size_type max_bytes = std::numeric_limits<size_type>::max(),
              typename Clock::duration ack_timeout = std::chrono::microseconds(0)) const
    {
        if (empty())
            return std::make_pair(-std::numeric_limits<double>::infinity(), ValueResult::EMPTY);
        else if (in_blackout(reference))
            return std::make_pair(-std::numeric_limits<double>::infinity(),
                                  ValueResult::IN_BLACKOUT);
        else if (top_size(reference, ack_timeout) > max_bytes)
            return std::make_pair(-std::numeric_limits<double>::infinity(),
                                  ValueResult::NEXT_MESSAGE_TOO_LARGE);
        else if (all_waiting_for_ack(reference, ack_timeout))
            return std::make_pair(-std::numeric_limits<double>::infinity(),
                                  ValueResult::ALL_MESSAGES_WAITING_FOR_ACK);
 
        using Duration = std::chrono::microseconds;
 
        double dt = std::chrono::duration_cast<Duration>(reference - last_access_).count();
        double ttl = goby::time::convert_duration<Duration>(cfg_.ttl_with_units()).count();
        double v_b = cfg_.value_base();
 
        double v = v_b * dt / ttl;
        return std::make_pair(v, ValueResult::VALUE_PROVIDED);
    }
 
    bool in_blackout(typename Clock::time_point reference = Clock::now()) const
    {
        auto blackout =
            goby::time::convert_duration<typename Clock::duration>(cfg_.blackout_time_with_units());
 
        return reference <= (last_access_ + blackout);
    }
    bool empty() const { return data_.empty(); }
 
    size_type size() const { return data_.size(); }
 
    void pop() { data_.pop_front(); }
 
    std::vector<Value> push(const T& t, typename Clock::time_point reference = Clock::now())
    {
        std::vector<Value> exceeded;
 
        if (cfg_.newest_first())
            data_.push_front(std::make_pair(zero_point_, Value({reference, t})));
        else
            data_.push_back(std::make_pair(zero_point_, Value({reference, t})));
 
        while (data_.size() > cfg_.max_queue())
        {
            exceeded.push_back(data_.back().second);
            data_.pop_back();
        }
        return exceeded;
    }
 
    std::vector<Value> expire(typename Clock::time_point reference = Clock::now())
    {
        std::vector<Value> expired;
 
        auto ttl = goby::time::convert_duration<typename Clock::duration>(cfg_.ttl_with_units());
        if (cfg_.newest_first())
        {
            while (!data_.empty() && reference > (data_.back().second.push_time + ttl))
            {
                expired.push_back(data_.back().second);
                data_.pop_back();
            }
        }
        else
        {
            while (!data_.empty() && reference > (data_.front().second.push_time + ttl))
            {
                expired.push_back(data_.front().second);
                data_.pop_front();
            }
        }
        return expired;
    }
 
    bool erase(const Value& value)
    {
        // start at the beginning as we are most likely to want to erase elements we recently asked for with top()
 
        for (auto it = data_.begin(), end = data_.end(); it != end; ++it)
        {
            const auto& datum_pair = it->second;
            if (datum_pair == value)
            {
                data_.erase(it);
                return true;
            }
 
            // if these are true, we're not going to find it so stop looking
            if (cfg_.newest_first() && datum_pair.push_time < value.push_time)
                break;
            else if (!cfg_.newest_first() && datum_pair.push_time > value.push_time)
                break;
        }
        return false;
    }
 
  private:
    goby::acomms::protobuf::DynamicBufferConfig cfg_;
 
    // pair of last send -> value
    std::deque<std::pair<typename Clock::time_point, Value>> data_;
    typename Clock::time_point last_access_{Clock::now()};
 
    typename Clock::time_point zero_point_{std::chrono::seconds(0)};
};
 
template <typename T, typename Clock = goby::time::SteadyClock> class DynamicBuffer
{
  public:
    DynamicBuffer() : DynamicBuffer(++count_) {}
    DynamicBuffer(int id)
    {
        glog_priority_group_ = "goby::acomms::buffer::priority::" + std::to_string(id);
        goby::glog.add_group(glog_priority_group_, util::Colors::yellow);
    }
    ~DynamicBuffer() {}
 
    using subbuffer_id_type = std::string;
    using size_type = typename DynamicSubBuffer<T, Clock>::size_type;
    using modem_id_type = int;
 
    struct Value
    {
        modem_id_type modem_id;
        subbuffer_id_type subbuffer_id;
        typename Clock::time_point push_time;
        T data;
    };
 
    void create(modem_id_type dest_id, const subbuffer_id_type& sub_id,
                const goby::acomms::protobuf::DynamicBufferConfig& cfg)
    {
        create(dest_id, sub_id, std::vector<goby::acomms::protobuf::DynamicBufferConfig>(1, cfg));
    }
 
    void create(modem_id_type dest_id, const subbuffer_id_type& sub_id,
                const std::vector<goby::acomms::protobuf::DynamicBufferConfig>& cfgs)
    {
        if (sub_.count(dest_id) && sub_.at(dest_id).count(sub_id))
            throw(goby::Exception("Subbuffer ID: " + sub_id + " already exists."));
 
        sub_[dest_id].insert(std::make_pair(sub_id, DynamicSubBuffer<T, Clock>(cfgs)));
    }
 
    void replace(modem_id_type dest_id, const subbuffer_id_type& sub_id,
                 const goby::acomms::protobuf::DynamicBufferConfig& cfg)
    {
        replace(dest_id, sub_id, std::vector<goby::acomms::protobuf::DynamicBufferConfig>(1, cfg));
    }
 
    void replace(modem_id_type dest_id, const subbuffer_id_type& sub_id,
                 const std::vector<goby::acomms::protobuf::DynamicBufferConfig>& cfgs)
    {
        remove(dest_id, sub_id);
        create(dest_id, sub_id, cfgs);
    }
 
    void update(modem_id_type dest_id, const subbuffer_id_type& sub_id,
                const goby::acomms::protobuf::DynamicBufferConfig& cfg)
    {
        update(dest_id, sub_id, std::vector<goby::acomms::protobuf::DynamicBufferConfig>(1, cfg));
    }
 
    void update(modem_id_type dest_id, const subbuffer_id_type& sub_id,
                const std::vector<goby::acomms::protobuf::DynamicBufferConfig>& cfgs)
    {
        auto it = sub_[dest_id].find(sub_id);
        if (it != sub_[dest_id].end())
            it->second.update(cfgs);
        else
            create(dest_id, sub_id, cfgs);
    }
 
    void remove(modem_id_type dest_id, const subbuffer_id_type& sub_id)
    {
        sub_[dest_id].erase(sub_id);
    }
 
    std::vector<Value> push(const Value& fvt)
    {
        std::vector<Value> exceeded;
        auto sub_exceeded = sub(fvt.modem_id, fvt.subbuffer_id).push(fvt.data, fvt.push_time);
        for (const auto& e : sub_exceeded)
            exceeded.push_back({fvt.modem_id, fvt.subbuffer_id, e.push_time, e.data});
        return exceeded;
    }
 
    bool empty() const
    {
        for (const auto& sub_id_p : sub_)
        {
            for (const auto& sub_p : sub_id_p.second)
            {
                if (!sub_p.second.empty())
                    return false;
            }
        }
 
        return true;
    }
 
    size_type size() const
    {
        size_type size = 0;
        for (const auto& sub_id_p : sub_)
        {
            for (const auto& sub_p : sub_id_p.second) size += sub_p.second.size();
        }
        return size;
    }
 
    Value top(modem_id_type dest_id = goby::acomms::QUERY_DESTINATION_ID,
              size_type max_bytes = std::numeric_limits<size_type>::max(),
              typename Clock::duration ack_timeout = std::chrono::microseconds(0))
    {
        using goby::glog;
 
        glog.is_debug1() &&
            glog << group(glog_priority_group_) << "Starting priority contest (dest: "
                 << (dest_id == goby::acomms::QUERY_DESTINATION_ID ? std::string("?")
                                                                   : std::to_string(dest_id))
                 << ", max_bytes: " << max_bytes << "):" << std::endl;
 
        typename std::unordered_map<subbuffer_id_type, DynamicSubBuffer<T, Clock>>::iterator
            winning_sub;
        double winning_value = -std::numeric_limits<double>::infinity();
 
        auto now = Clock::now();
 
        if (dest_id != goby::acomms::QUERY_DESTINATION_ID && !sub_.count(dest_id))
            throw(DynamicBufferNoDataException());
 
        // if QUERY_DESTINATION_ID, search all subbuffers, otherwise just search the ones that were specified by dest_id
        for (auto sub_id_it = (dest_id == goby::acomms::QUERY_DESTINATION_ID) ? sub_.begin()
                                                                              : sub_.find(dest_id),
                  sub_id_end = (dest_id == goby::acomms::QUERY_DESTINATION_ID)
                                   ? sub_.end()
                                   : ++sub_.find(dest_id);
             sub_id_it != sub_id_end; ++sub_id_it)
        {
            for (auto sub_it = sub_id_it->second.begin(), sub_end = sub_id_it->second.end();
                 sub_it != sub_end; ++sub_it)
            {
                double value;
                typename DynamicSubBuffer<T, Clock>::ValueResult result;
                std::tie(value, result) = sub_it->second.top_value(now, max_bytes, ack_timeout);
 
                std::string value_or_reason;
                switch (result)
                {
                    case DynamicSubBuffer<T, Clock>::ValueResult::VALUE_PROVIDED:
                        value_or_reason = std::to_string(value);
                        break;
 
                    case DynamicSubBuffer<T, Clock>::ValueResult::EMPTY:
                        value_or_reason = "empty";
                        break;
 
                    case DynamicSubBuffer<T, Clock>::ValueResult::IN_BLACKOUT:
                        value_or_reason = "blackout";
                        break;
 
                    case DynamicSubBuffer<T, Clock>::ValueResult::NEXT_MESSAGE_TOO_LARGE:
                        value_or_reason = "too large";
                        break;
 
                    case DynamicSubBuffer<T, Clock>::ValueResult::ALL_MESSAGES_WAITING_FOR_ACK:
                        value_or_reason = "ack wait";
                        break;
                }
 
                glog.is_debug1() && glog << group(glog_priority_group_) << "\t" << sub_it->first
                                         << " [dest: " << sub_id_it->first
                                         << ", n: " << sub_it->second.size()
                                         << "]: " << value_or_reason << std::endl;
 
                if (value > winning_value)
                {
                    winning_value = value;
                    winning_sub = sub_it;
                    dest_id = sub_id_it->first;
                }
            }
        }
 
        if (winning_value == -std::numeric_limits<double>::infinity())
            throw(DynamicBufferNoDataException());
 
        const auto& top_p = winning_sub->second.top(now, ack_timeout);
        glog.is_debug1() && glog << group(glog_priority_group_) << "Winner: " << winning_sub->first
                                 << " (" << data_size(top_p.data) << "B)" << std::endl;
 
        return {dest_id, winning_sub->first, top_p.push_time, top_p.data};
    }
 
    bool erase(const Value& value)
    {
        return sub(value.modem_id, value.subbuffer_id).erase({value.push_time, value.data});
    }
 
    std::vector<Value> expire()
    {
        auto now = Clock::now();
        std::vector<Value> expired;
        for (auto& sub_id_p : sub_)
        {
            for (auto& sub_p : sub_id_p.second)
            {
                auto sub_expired = sub_p.second.expire(now);
                for (const auto& e : sub_expired)
                    expired.push_back({sub_id_p.first, sub_p.first, e.push_time, e.data});
            }
        }
        return expired;
    }
 
    DynamicSubBuffer<T, Clock>& sub(modem_id_type dest_id, const subbuffer_id_type& sub_id)
    {
        if (!sub_.count(dest_id) || !sub_.at(dest_id).count(sub_id))
            throw(goby::Exception("Subbuffer ID: " + sub_id +
                                  " does not exist, must call create(...) first."));
        return sub_.at(dest_id).at(sub_id);
    }
 
  private:
    // destination -> subbuffer id (group/type) -> subbuffer
    std::map<modem_id_type, std::unordered_map<subbuffer_id_type, DynamicSubBuffer<T, Clock>>> sub_;
 
    std::string glog_priority_group_;
    static std::atomic<int> count_;
 
}; // namespace acomms
 
template <typename T, typename Clock> std::atomic<int> DynamicBuffer<T, Clock>::count_(0);
 
} // namespace acomms
} // namespace goby
 
#endif