#include #include #include #include #include #include #include #include namespace vanetza { namespace geonet { CbfPacket::CbfPacket(PendingPacket&& packet, const MacAddress& sender) : m_packet(std::move(packet)), m_sender(sender) { } CbfPacket::CbfPacket(PendingPacket&& packet, const MacAddress& sender) : m_packet(PendingPacket(std::move(packet), cBroadcastMacAddress)), m_sender(sender) { } const MacAddress& CbfPacket::sender() const { return m_sender; } const Address& CbfPacket::source() const { return m_packet.pdu().extended().source_position.gn_addr; } SequenceNumber CbfPacket::sequence_number() const { return m_packet.pdu().extended().sequence_number; } Clock::duration CbfPacket::reduce_lifetime(Clock::duration d) { return m_packet.reduce_lifetime(d); } std::size_t CbfPacket::length() const { return m_packet.length(); } CbfPacketBuffer::CbfPacketBuffer(Runtime& rt, TimerCallback cb, std::unique_ptr cnt, std::size_t bytes) : m_runtime(rt), m_counter(std::move(cnt)), m_capacity_bytes(bytes), m_stored_bytes(0), m_timer_callback(cb) { } CbfPacketBuffer::~CbfPacketBuffer() { m_runtime.cancel(this); } bool CbfPacketBuffer::remove(const Identifier& id) { bool packet_dropped = false; auto& id_map = m_timers.right; auto found = id_map.find(id); if (found != id_map.end()) { auto& packet = found->info; m_stored_bytes -= packet->length(); m_counter->remove(id); m_packets.erase(packet); remove_timer(m_timers.project_left(found)); packet_dropped = true; } assert(m_packets.size() == m_timers.size()); return packet_dropped; } void CbfPacketBuffer::remove_timer(typename timer_bimap::left_map::iterator timer_it) { auto& timer_map = m_timers.left; auto successor = timer_map.erase(timer_it); if (successor == timer_map.begin() && !timer_map.empty()) { // erased timer was scheduled one, reschedule timer trigger schedule_timer(); } } void CbfPacketBuffer::add(CbfPacket&& packet, Clock::duration timeout) { if(timeout <= Clock::duration::zero()) return; m_stored_bytes += packet.length(); const auto first_timer = m_timers.left.begin(); // do head drop if necessary while (m_stored_bytes > m_capacity_bytes && !m_packets.empty()) { m_stored_bytes -= m_packets.front().length(); const auto id = identifier(m_packets.front()); m_timers.right.erase(id); m_counter->remove(id); m_packets.pop_front(); } Timer timer = { m_runtime, timeout }; const Identifier id = identifier(packet); m_packets.emplace_back(std::move(packet)); using timer_value = timer_bimap::value_type; auto insertion = m_timers.insert(timer_value { timer, id, std::prev(m_packets.end()) }); if (!insertion.second) { m_stored_bytes -= m_packets.back().length(); m_packets.pop_back(); } else { m_counter->add(id); } // first expirying timer has changed (head drop or added packet) if (m_timers.left.begin() != first_timer) { schedule_timer(); } assert(m_packets.size() == m_timers.size()); } void CbfPacketBuffer::update(const Identifier& id, Clock::duration timeout) { auto& id_map = m_timers.right; auto found = id_map.find(id); if (found != id_map.end()) { const Timer& timer = found->second; CbfPacket& cbf_packet = *found->info; reduce_lifetime(timer, cbf_packet); id_map.replace_data(found, Timer { m_runtime, timeout}); m_counter->increment(id); } } boost::optional CbfPacketBuffer::fetch(const Identifier& id) { boost::optional packet; auto& id_map = m_timers.right; auto found = id_map.find(id); if (found != id_map.end()) { const Timer& timer = found->second; CbfPacket& cbf_packet = *found->info; bool valid_packet = reduce_lifetime(timer, cbf_packet); m_stored_bytes -= cbf_packet.length(); if (valid_packet) { packet.emplace(std::move(cbf_packet)); } m_counter->remove(id); m_packets.erase(found->info); remove_timer(m_timers.project_left(found)); } return packet; } const CbfPacket* CbfPacketBuffer::find(const Identifier& id) const { const auto& id_map = m_timers.right; auto found = id_map.find(id); return found != id_map.end() ? &(*found->info) : nullptr; } std::size_t CbfPacketBuffer::counter(const Identifier& id) const { return m_counter->counter(id); } void CbfPacketBuffer::flush() { // fetch all expired timers const Timer now { m_runtime, std::chrono::seconds(0) }; auto end = m_timers.left.upper_bound(now); for (auto it = m_timers.left.begin(); it != end;) { // reduce LT by queuing time const Timer& timer = it->first; CbfPacket& packet = *it->info; bool valid_packet = reduce_lifetime(timer, packet); m_stored_bytes -= packet.length(); if (valid_packet) { m_timer_callback(std::move(packet).packet()); } m_counter->remove(it->second); m_packets.erase(it->info); it = m_timers.left.erase(it); } // schedule timer if not empty if (!m_timers.empty()) { schedule_timer(); } } bool CbfPacketBuffer::reduce_lifetime(const Timer& timer, CbfPacket& packet) const { const auto queuing_time = m_runtime.now() - timer.start; return packet.reduce_lifetime(queuing_time) > Clock::duration::zero(); } void CbfPacketBuffer::schedule_timer() { assert(!m_timers.empty()); m_runtime.cancel(this); Runtime::Callback cb = [this](Clock::time_point) { flush(); }; m_runtime.schedule(m_timers.left.begin()->first.expiry, cb, this); } CbfPacketBuffer::Timer::Timer(const Runtime& rt, Clock::duration timeout) : expiry(rt.now() + timeout), start(rt.now()) { } bool CbfPacketBuffer::Timer::operator<(const Timer& other) const { return this->expiry < other.expiry; } } // namespace geonet } // namespace vanetza