Digital_Voice_Modem_Project
/
dvmhost
mirror of https://github.com/DVMProject/dvmhost.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

EXPERIMENTAL: implement new stream rejection logic when a call collision occurs, this is smarter then the existing logic in that a stream will fail processing validation and be rejected once identified as a collider;

Browse Source
pull/121/merge
Bryan Biedenkapp 3 weeks ago
parent d29ce3c74f
commit 8784e06e3b
9 changed files with 206 additions and 4 deletions
Show Stats Download Patch File Download Diff File

1
src/fne/network/TrafficNetwork.h
Unescape View File

@ -93,6 +93,7 @@ namespace network
#define INFLUXDB_ERRSTR_INV_SLOT "invalid slot for talkgroup"
#define INFLUXDB_ERRSTR_RID_NOT_PERMITTED "RID not permitted for talkgroup"
#define INFLUXDB_ERRSTR_ILLEGAL_RID_ACCESS "illegal/unknown RID attempted access"
#define INFLUXDB_ERRSTR_CALL_NOT_PERMITTED "call not permitted for talkgroup"
// ---------------------------------------------------------------------------
// Class Prototypes

49
src/fne/network/callhandler/TagAnalogData.cpp
Unescape View File

@ -43,6 +43,7 @@ TagAnalogData::TagAnalogData(TrafficNetwork* network, bool debug) :
m_lastParrotSrcId(0U),
m_lastParrotDstId(0U),
m_status(),
m_rejectedCallStreams(),
m_debug(debug)
{
assert(network != nullptr);
@ -130,6 +131,11 @@ bool TagAnalogData::processFrame(const uint8_t* data, uint32_t len, uint32_t pee
else if (!fromUpstream)
LogInfoEx(LOG_MASTER, CALL_END_LOG);
// clear any rejected call streams for this TG
m_rejectedCallStreams.lock(false);
m_rejectedCallStreams[dstId].clear();
m_rejectedCallStreams.unlock();
if (!tg.config().parrot()) {
m_network->m_totalActiveCalls--;
if (m_network->m_totalActiveCalls < 0)
@ -217,11 +223,21 @@ bool TagAnalogData::processFrame(const uint8_t* data, uint32_t len, uint32_t pee
m_status[dstId].srcId = srcId;
m_status[dstId].ssrc = ssrc;
m_status.unlock();
// because the call stream source has reset, clear any rejected call streams for
// this TG to allow the new source to transmit
m_rejectedCallStreams.lock(false);
m_rejectedCallStreams[dstId].clear();
m_rejectedCallStreams.unlock();
}
else {
LogWarning((fromUpstream) ? LOG_PEER : LOG_MASTER, "Analog, Call Collision, peer = %u, ssrc = %u, srcId = %u, dstId = %u, streamId = %u, rxPeer = %u, rxSrcId = %u, rxDstId = %u, rxStreamId = %u, fromUpstream = %u",
peerId, ssrc, srcId, dstId, streamId, status.peerId, status.srcId, status.dstId, status.streamId, fromUpstream);
m_rejectedCallStreams.lock(false);
m_rejectedCallStreams[dstId].push_back(streamId);
m_rejectedCallStreams.unlock();
m_network->m_totalCallCollisions++;
return false;
@ -277,6 +293,11 @@ bool TagAnalogData::processFrame(const uint8_t* data, uint32_t len, uint32_t pee
m_status[dstId].activeCall = true;
m_status.unlock();
// clear any rejected call streams for this TG
m_rejectedCallStreams.lock(false);
m_rejectedCallStreams[dstId].clear();
m_rejectedCallStreams.unlock();
if (!tg.config().parrot()) {
m_network->m_totalCallsProcessed++;
m_network->m_totalActiveCalls++;
@ -705,6 +726,34 @@ bool TagAnalogData::validate(uint32_t peerId, data::NetData& data, uint32_t stre
if (data.getFrameType() == AudioFrameType::TERMINATOR)
return true;
// is the call stream rejected?
m_rejectedCallStreams.lock(false);
std::vector<uint32_t> rejectedStreams = m_rejectedCallStreams[data.getDstId()];
if (std::find(rejectedStreams.begin(), rejectedStreams.end(), streamId) != rejectedStreams.end()) {
// report error event to InfluxDB
if (m_network->m_enableInfluxDB) {
influxdb::QueryBuilder()
.meas("call_error_event")
.tag("peerId", std::to_string(peerId))
.tag("streamId", std::to_string(streamId))
.tag("srcId", std::to_string(data.getSrcId()))
.tag("dstId", std::to_string(data.getDstId()))
.field("message", std::string(INFLUXDB_ERRSTR_CALL_NOT_PERMITTED))
.timestamp(std::chrono::duration_cast<std::chrono::nanoseconds>(std::chrono::system_clock::now().time_since_epoch()).count())
.requestAsync(m_network->m_influxServer);
}
if (m_network->m_logDenials)
LogError(LOG_ANALOG, INFLUXDB_ERRSTR_CALL_NOT_PERMITTED ", peer = %u, srcId = %u, dstId = %u", peerId, data.getSrcId(), data.getDstId());
m_rejectedCallStreams.unlock();
// report In-Call Control to the peer sending traffic
m_network->writePeerICC(peerId, streamId, NET_SUBFUNC::PROTOCOL_SUBFUNC_ANALOG, NET_ICC::REJECT_TRAFFIC, data.getDstId());
return false;
}
m_rejectedCallStreams.unlock();
// is this a private call?
if (!data.getGroup()) {
// is the destination ID a blacklisted ID?

3
src/fne/network/callhandler/TagAnalogData.h
Unescape View File

@ -4,7 +4,7 @@
* GPLv2 Open Source. Use is subject to license terms.
* DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
*
* Copyright (C) 2025 Bryan Biedenkapp, N2PLL
* Copyright (C) 2025-2026 Bryan Biedenkapp, N2PLL
*
*/
/**
@ -203,6 +203,7 @@ namespace network
};
typedef std::pair<const uint32_t, RxStatus> StatusMapPair;
concurrent::unordered_map<uint32_t, RxStatus> m_status;
concurrent::unordered_map<uint32_t, std::vector<uint32_t>> m_rejectedCallStreams;
bool m_debug;

50
src/fne/network/callhandler/TagDMRData.cpp
Unescape View File

@ -48,6 +48,7 @@ TagDMRData::TagDMRData(TrafficNetwork* network, bool debug) :
m_lastParrotDstId(0U),
m_status(),
m_statusPVCall(),
m_rejectedCallStreams(),
m_debug(debug)
{
assert(network != nullptr);
@ -213,6 +214,11 @@ bool TagDMRData::processFrame(const uint8_t* data, uint32_t len, uint32_t peerId
LogInfoEx(LOG_MASTER, CALL_END_LOG);
}
// clear any rejected call streams for this TG
m_rejectedCallStreams.lock(false);
m_rejectedCallStreams[dstId].clear();
m_rejectedCallStreams.unlock();
if (!tg.config().parrot()) {
m_network->m_totalActiveCalls--;
if (m_network->m_totalActiveCalls < 0)
@ -329,10 +335,20 @@ bool TagDMRData::processFrame(const uint8_t* data, uint32_t len, uint32_t peerId
m_status[dstId].srcId = srcId;
m_status[dstId].ssrc = ssrc;
m_status.unlock();
// because the call stream source has reset, clear any rejected call streams for
// this TG to allow the new source to transmit
m_rejectedCallStreams.lock(false);
m_rejectedCallStreams[dstId].clear();
m_rejectedCallStreams.unlock();
} else {
LogWarning((fromUpstream) ? LOG_PEER : LOG_MASTER, "DMR, Call Collision, peer = %u, ssrc = %u, srcId = %u, dstId = %u, slotNo = %u, streamId = %u, rxPeer = %u, rxSrcId = %u, rxDstId = %u, rxSlotNo = %u, rxStreamId = %u, fromUpstream = %u",
peerId, ssrc, srcId, dstId, slotNo, streamId, status.peerId, status.srcId, status.dstId, status.slotNo, status.streamId, fromUpstream);
m_rejectedCallStreams.lock(false);
m_rejectedCallStreams[dstId].push_back(streamId);
m_rejectedCallStreams.unlock();
m_network->m_totalCallCollisions++;
return false;
@ -391,6 +407,11 @@ bool TagDMRData::processFrame(const uint8_t* data, uint32_t len, uint32_t peerId
m_status[dstId].activeCall = true;
m_status.unlock();
// clear any rejected call streams for this TG
m_rejectedCallStreams.lock(false);
m_rejectedCallStreams[dstId].clear();
m_rejectedCallStreams.unlock();
if (!tg.config().parrot()) {
m_network->m_totalCallsProcessed++;
m_network->m_totalActiveCalls++;
@ -1192,6 +1213,35 @@ bool TagDMRData::validate(uint32_t peerId, data::NetData& data, lc::CSBK* csbk,
return true;
}
// is the call stream rejected?
m_rejectedCallStreams.lock(false);
std::vector<uint32_t> rejectedStreams = m_rejectedCallStreams[data.getDstId()];
if (std::find(rejectedStreams.begin(), rejectedStreams.end(), streamId) != rejectedStreams.end()) {
// report error event to InfluxDB
if (m_network->m_enableInfluxDB) {
influxdb::QueryBuilder()
.meas("call_error_event")
.tag("peerId", std::to_string(peerId))
.tag("streamId", std::to_string(streamId))
.tag("srcId", std::to_string(data.getSrcId()))
.tag("dstId", std::to_string(data.getDstId()))
.field("message", std::string(INFLUXDB_ERRSTR_CALL_NOT_PERMITTED))
.field("slot", std::to_string(data.getSlotNo()))
.timestamp(std::chrono::duration_cast<std::chrono::nanoseconds>(std::chrono::system_clock::now().time_since_epoch()).count())
.requestAsync(m_network->m_influxServer);
}
if (m_network->m_logDenials)
LogError(LOG_DMR, "DMR Slot %u, " INFLUXDB_ERRSTR_CALL_NOT_PERMITTED ", peer = %u, srcId = %u, dstId = %u", data.getSlotNo(), peerId, data.getSrcId(), data.getDstId());
m_rejectedCallStreams.unlock();
// report In-Call Control to the peer sending traffic
m_network->writePeerICC(peerId, streamId, NET_SUBFUNC::PROTOCOL_SUBFUNC_DMR, NET_ICC::REJECT_TRAFFIC, data.getDstId(), data.getSlotNo());
return false;
}
m_rejectedCallStreams.unlock();
// is this a private call?
if (data.getFLCO() == FLCO::PRIVATE) {
// is the destination ID a blacklisted ID?

3
src/fne/network/callhandler/TagDMRData.h
Unescape View File

@ -4,7 +4,7 @@
* GPLv2 Open Source. Use is subject to license terms.
* DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
*
* Copyright (C) 2023-2024 Bryan Biedenkapp, N2PLL
* Copyright (C) 2023-2026 Bryan Biedenkapp, N2PLL
*
*/
/**
@ -254,6 +254,7 @@ namespace network
typedef std::pair<const uint32_t, RxStatus> StatusMapPair;
concurrent::unordered_map<uint32_t, RxStatus> m_status;
concurrent::unordered_map<uint32_t, RxStatus> m_statusPVCall;
concurrent::unordered_map<uint32_t, std::vector<uint32_t>> m_rejectedCallStreams;
friend class packetdata::DMRPacketData;
packetdata::DMRPacketData* m_packetData;

49
src/fne/network/callhandler/TagNXDNData.cpp
Unescape View File

@ -51,6 +51,7 @@ TagNXDNData::TagNXDNData(TrafficNetwork* network, bool debug) :
m_lastParrotDstId(0U),
m_status(),
m_statusPVCall(),
m_rejectedCallStreams(),
m_debug(debug)
{
assert(network != nullptr);
@ -252,6 +253,11 @@ bool TagNXDNData::processFrame(const uint8_t* data, uint32_t len, uint32_t peerI
LogInfoEx(LOG_MASTER, CALL_END_LOG);
}
// clear any rejected call streams for this TG
m_rejectedCallStreams.lock(false);
m_rejectedCallStreams[dstId].clear();
m_rejectedCallStreams.unlock();
if (!tg.config().parrot()) {
m_network->m_totalActiveCalls--;
if (m_network->m_totalActiveCalls < 0)
@ -357,10 +363,20 @@ bool TagNXDNData::processFrame(const uint8_t* data, uint32_t len, uint32_t peerI
m_status[dstId].srcId = srcId;
m_status[dstId].ssrc = ssrc;
m_status.unlock();
// because the call stream source has reset, clear any rejected call streams for
// this TG to allow the new source to transmit
m_rejectedCallStreams.lock(false);
m_rejectedCallStreams[dstId].clear();
m_rejectedCallStreams.unlock();
} else {
LogWarning((fromUpstream) ? LOG_PEER : LOG_MASTER, "NXDN, Call Collision, peer = %u, ssrc = %u, srcId = %u, dstId = %u, streamId = %u, rxPeer = %u, rxSrcId = %u, rxDstId = %u, rxStreamId = %u, fromUpstream = %u",
peerId, ssrc, srcId, dstId, streamId, status.peerId, status.srcId, status.dstId, status.streamId, fromUpstream);
m_rejectedCallStreams.lock(false);
m_rejectedCallStreams[dstId].push_back(streamId);
m_rejectedCallStreams.unlock();
m_network->m_totalCallCollisions++;
return false;
@ -417,6 +433,11 @@ bool TagNXDNData::processFrame(const uint8_t* data, uint32_t len, uint32_t peerI
m_status[dstId].activeCall = true;
m_status.unlock();
// clear any rejected call streams for this TG
m_rejectedCallStreams.lock(false);
m_rejectedCallStreams[dstId].clear();
m_rejectedCallStreams.unlock();
if (!tg.config().parrot()) {
m_network->m_totalCallsProcessed++;
m_network->m_totalActiveCalls++;
@ -1004,6 +1025,34 @@ bool TagNXDNData::validate(uint32_t peerId, lc::RTCH& lc, uint8_t messageType, u
if (messageType == MessageType::RTCH_TX_REL || messageType == MessageType::RTCH_TX_REL_EX)
return true;
// is the call stream rejected?
m_rejectedCallStreams.lock(false);
std::vector<uint32_t> rejectedStreams = m_rejectedCallStreams[lc.getDstId()];
if (std::find(rejectedStreams.begin(), rejectedStreams.end(), streamId) != rejectedStreams.end()) {
// report error event to InfluxDB
if (m_network->m_enableInfluxDB) {
influxdb::QueryBuilder()
.meas("call_error_event")
.tag("peerId", std::to_string(peerId))
.tag("streamId", std::to_string(streamId))
.tag("srcId", std::to_string(lc.getSrcId()))
.tag("dstId", std::to_string(lc.getDstId()))
.field("message", std::string(INFLUXDB_ERRSTR_CALL_NOT_PERMITTED))
.timestamp(std::chrono::duration_cast<std::chrono::nanoseconds>(std::chrono::system_clock::now().time_since_epoch()).count())
.requestAsync(m_network->m_influxServer);
}
if (m_network->m_logDenials)
LogError(LOG_NXDN, INFLUXDB_ERRSTR_CALL_NOT_PERMITTED ", peer = %u, srcId = %u, dstId = %u", peerId, lc.getSrcId(), lc.getDstId());
m_rejectedCallStreams.unlock();
// report In-Call Control to the peer sending traffic
m_network->writePeerICC(peerId, streamId, NET_SUBFUNC::PROTOCOL_SUBFUNC_NXDN, NET_ICC::REJECT_TRAFFIC, lc.getDstId());
return false;
}
m_rejectedCallStreams.unlock();
// is this a private call?
if (!lc.getGroup()) {
// is the destination ID a blacklisted ID?

3
src/fne/network/callhandler/TagNXDNData.h
Unescape View File

@ -4,7 +4,7 @@
* GPLv2 Open Source. Use is subject to license terms.
* DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
*
* Copyright (C) 2023-2024 Bryan Biedenkapp, N2PLL
* Copyright (C) 2023-2026 Bryan Biedenkapp, N2PLL
*
*/
/**
@ -218,6 +218,7 @@ namespace network
typedef std::pair<const uint32_t, RxStatus> StatusMapPair;
concurrent::unordered_map<uint32_t, RxStatus> m_status;
concurrent::unordered_map<uint32_t, RxStatus> m_statusPVCall;
concurrent::unordered_map<uint32_t, std::vector<uint32_t>> m_rejectedCallStreams;
bool m_debug;

49
src/fne/network/callhandler/TagP25Data.cpp
Unescape View File

@ -48,6 +48,7 @@ TagP25Data::TagP25Data(TrafficNetwork* network, bool debug) :
m_lastParrotDstId(0U),
m_status(),
m_statusPVCall(),
m_rejectedCallStreams(),
m_packetData(nullptr),
m_debug(debug)
{
@ -320,6 +321,11 @@ bool TagP25Data::processFrame(const uint8_t* data, uint32_t len, uint32_t peerId
LogInfoEx(LOG_MASTER, CALL_END_LOG);
}
// clear any rejected call streams for this TG
m_rejectedCallStreams.lock(false);
m_rejectedCallStreams[dstId].clear();
m_rejectedCallStreams.unlock();
if (!tg.config().parrot()) {
m_network->m_totalActiveCalls--;
if (m_network->m_totalActiveCalls < 0)
@ -435,11 +441,21 @@ bool TagP25Data::processFrame(const uint8_t* data, uint32_t len, uint32_t peerId
m_status[dstId].srcId = srcId;
m_status[dstId].ssrc = ssrc;
m_status.unlock();
// because the call stream source has reset, clear any rejected call streams for
// this TG to allow the new source to transmit
m_rejectedCallStreams.lock(false);
m_rejectedCallStreams[dstId].clear();
m_rejectedCallStreams.unlock();
}
else {
LogWarning((fromUpstream) ? LOG_PEER : LOG_MASTER, "P25, Call Collision, peer = %u, ssrc = %u, sysId = $%03X, netId = $%05X, srcId = %u, dstId = %u, streamId = %u, rxPeer = %u, rxSrcId = %u, rxDstId = %u, rxStreamId = %u, fromUpstream = %u",
peerId, ssrc, sysId, netId, srcId, dstId, streamId, status.peerId, status.srcId, status.dstId, status.streamId, fromUpstream);
m_rejectedCallStreams.lock(false);
m_rejectedCallStreams[dstId].push_back(streamId);
m_rejectedCallStreams.unlock();
m_network->m_totalCallCollisions++;
return false;
@ -496,6 +512,11 @@ bool TagP25Data::processFrame(const uint8_t* data, uint32_t len, uint32_t peerId
m_status[dstId].activeCall = true;
m_status.unlock();
// clear any rejected call streams for this TG
m_rejectedCallStreams.lock(false);
m_rejectedCallStreams[dstId].clear();
m_rejectedCallStreams.unlock();
if (!tg.config().parrot()) {
m_network->m_totalCallsProcessed++;
m_network->m_totalActiveCalls++;
@ -1645,6 +1666,34 @@ bool TagP25Data::validate(uint32_t peerId, lc::LC& control, DUID::E duid, const
return true;
}
// is the call stream rejected?
m_rejectedCallStreams.lock(false);
std::vector<uint32_t> rejectedStreams = m_rejectedCallStreams[control.getDstId()];
if (std::find(rejectedStreams.begin(), rejectedStreams.end(), streamId) != rejectedStreams.end()) {
// report error event to InfluxDB
if (m_network->m_enableInfluxDB) {
influxdb::QueryBuilder()
.meas("call_error_event")
.tag("peerId", std::to_string(peerId))
.tag("streamId", std::to_string(streamId))
.tag("srcId", std::to_string(control.getSrcId()))
.tag("dstId", std::to_string(control.getDstId()))
.field("message", std::string(INFLUXDB_ERRSTR_CALL_NOT_PERMITTED))
.timestamp(std::chrono::duration_cast<std::chrono::nanoseconds>(std::chrono::system_clock::now().time_since_epoch()).count())
.requestAsync(m_network->m_influxServer);
}
if (m_network->m_logDenials)
LogError(LOG_P25, INFLUXDB_ERRSTR_CALL_NOT_PERMITTED ", peer = %u, srcId = %u, dstId = %u", peerId, control.getSrcId(), control.getDstId());
m_rejectedCallStreams.unlock();
// report In-Call Control to the peer sending traffic
m_network->writePeerICC(peerId, streamId, NET_SUBFUNC::PROTOCOL_SUBFUNC_P25, NET_ICC::REJECT_TRAFFIC, control.getDstId());
return false;
}
m_rejectedCallStreams.unlock();
// validate private call in-progress
bool privateCallInProgress = false;
if ((control.getLCO() != LCO::PRIVATE) && !control.getGroup()) {

3
src/fne/network/callhandler/TagP25Data.h
Unescape View File

@ -4,7 +4,7 @@
* GPLv2 Open Source. Use is subject to license terms.
* DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
*
* Copyright (C) 2023-2024 Bryan Biedenkapp, N2PLL
* Copyright (C) 2023-2026 Bryan Biedenkapp, N2PLL
*
*/
/**
@ -267,6 +267,7 @@ namespace network
typedef std::pair<const uint32_t, RxStatus> StatusMapPair;
concurrent::unordered_map<uint32_t, RxStatus> m_status;
concurrent::unordered_map<uint32_t, RxStatus> m_statusPVCall;
concurrent::unordered_map<uint32_t, std::vector<uint32_t>> m_rejectedCallStreams;
friend class packetdata::P25PacketData;
packetdata::P25PacketData* m_packetData;

Write Preview
Loading…
Cancel
Save

Powered by TurnKey Linux.