#include "PrtCnvRadioTrafficThread.hpp" #include "PrtCnvDbThread.hpp" #include "PrtCnvThreadStarter.hpp" #include #include #include #include namespace PrtCnv { PrtCnvRadioTrafficThread::PrtCnvRadioTrafficThread(PrtCnvThreadStarter* a_parent) : PrtCnvThread(a_parent, "PrtCnvRadioTrafficThread") , m_receiveLoad(PrtCnvRadioThread::E_TrafficLoad::Unknown) , m_fullLoad(100) , m_highLoad(50) , m_hystLoad(5) , m_calcTimeout(1000) , m_cleanupTimeout(5000) , m_mutex(QMutex::Recursive) {} void PrtCnvRadioTrafficThread::newTelegramArrived(unsigned int size) { time_t curTime = time(nullptr); QMutexLocker l(&m_mutex); DEBUG_LOG(m_logger) << "newTelegramArrived with size " << size << " at " << curTime; m_data[curTime] += size; m_telegrams[curTime]++; } int32_t PrtCnvRadioTrafficThread::start() { const QString trafficControlGroup(QStringLiteral("TrafficControl")); int highLoad = m_Config->getInteger(trafficControlGroup, QStringLiteral("ReceiveQueueHighLoad"), 60); int fullLoad = m_Config->getInteger(trafficControlGroup, QStringLiteral("ReceiveQueueFullLoad"), 80); int hystLoad = m_Config->getInteger(trafficControlGroup, QStringLiteral("ReceiveQueueHysteresis"), 5); int avgTime = m_Config->getInteger(trafficControlGroup, QStringLiteral("ReceiveAveragingTime"), 5); float trafficPerReceiverKB = m_Config->getFloat(trafficControlGroup, QStringLiteral("ReceiveThroughput"), 3.6f); int cntTransceiver = m_Config->getInteger(trafficControlGroup, QStringLiteral("TransceiverCount"), 6); if (highLoad < 0 || highLoad > 100) highLoad = 60; if (fullLoad < highLoad || fullLoad > 100) fullLoad = 80; if (hystLoad < 0 || hystLoad > 10) hystLoad = 5; if (avgTime <= 0 || avgTime > 60) avgTime = 5; if (trafficPerReceiverKB < 1 || trafficPerReceiverKB > 1000) trafficPerReceiverKB = 3.6f; if (cntTransceiver < 1 || cntTransceiver > 100) cntTransceiver = 6; float maxTrafficBytes = trafficPerReceiverKB * cntTransceiver * 1024; m_fullLoad = maxTrafficBytes * fullLoad / 100.f; m_highLoad = maxTrafficBytes * highLoad / 100.f; m_hystLoad = maxTrafficBytes * hystLoad / 100.f; m_cleanupTimeout = avgTime * 1000; DEBUG_LOG(m_logger) << "Read config ReceiveQueueHighLoad: " << highLoad; DEBUG_LOG(m_logger) << "Read config ReceiveQueueFullLoad: " << fullLoad; DEBUG_LOG(m_logger) << "Read config ReceiveQueueHysteresis: " << hystLoad; DEBUG_LOG(m_logger) << "Read config ReceiveAveragingTime: " << avgTime; DEBUG_LOG(m_logger) << "Read config ReceiveThroughput: " << trafficPerReceiverKB; DEBUG_LOG(m_logger) << "Read config TransceiverCount: " << cntTransceiver; INFO_LOG(m_logger) << "Using to following limits: high - " << m_highLoad << " full - " << m_fullLoad << " hyst - " << m_hystLoad << " avg. time - " << m_cleanupTimeout / 1000; QTimer::singleShot(m_calcTimeout, this, &PrtCnvRadioTrafficThread::onCalculateTrafficLoad); QTimer::singleShot(m_cleanupTimeout, this, &PrtCnvRadioTrafficThread::onCleanupTimeout); return 0; } void PrtCnvRadioTrafficThread::onCalculateTrafficLoad() { const time_t curTime = time(nullptr); unsigned int payload = 0; { QMutexLocker l(&m_mutex); for (int i = 1; i <= m_cleanupTimeout / 1000; ++i) payload += m_data.value(curTime - i); } payload /= (m_cleanupTimeout / 1000); INFO_LOG(m_logger) << "Average Payload in the last " << m_cleanupTimeout << " Milliseconds: " << payload << " Bytes/sec, last sec: " << m_data[curTime - 1] << " Bytes, #telegrams: " << m_telegrams[curTime - 1]; PrtCnvRadioThread::E_TrafficLoad newReceiveLoad; if (payload >= m_fullLoad) { newReceiveLoad = PrtCnvRadioThread::E_TrafficLoad::FullLoad; if (m_receiveLoad == PrtCnvRadioThread::E_TrafficLoad::HighLoad) { if(payload < m_fullLoad + m_hystLoad) newReceiveLoad = PrtCnvRadioThread::E_TrafficLoad::HighLoad; } } else if (payload >= m_highLoad) { newReceiveLoad = PrtCnvRadioThread::E_TrafficLoad::HighLoad; if (m_receiveLoad == PrtCnvRadioThread::E_TrafficLoad::NormalLoad) { if (payload < m_highLoad + m_hystLoad) newReceiveLoad = PrtCnvRadioThread::E_TrafficLoad::NormalLoad; } if (m_receiveLoad == PrtCnvRadioThread::E_TrafficLoad::FullLoad) { if (payload >= m_fullLoad - m_hystLoad) newReceiveLoad = PrtCnvRadioThread::E_TrafficLoad::FullLoad; } } else { newReceiveLoad = PrtCnvRadioThread::E_TrafficLoad::NormalLoad; if (m_receiveLoad == PrtCnvRadioThread::E_TrafficLoad::HighLoad) { if(payload >= m_highLoad - m_hystLoad) newReceiveLoad = PrtCnvRadioThread::E_TrafficLoad::HighLoad; } } if (m_receiveLoad != newReceiveLoad) { m_receiveLoad = newReceiveLoad; unsigned int wartezeit_posmldg = (static_cast(m_receiveLoad) * 10) + 1; INFO_LOG(m_logger) << "New traffic load: " << m_receiveLoad; PrtCnvDbThread *dbThread = qobject_cast(m_Parent->getOtherThread(PrtCnvApplication::e_Db)); if (dbThread) { dbThread->setAirLoad(m_receiveLoad); dbThread->setPositionSendTime(wartezeit_posmldg); Q_ASSERT(QThread::currentThreadId() == m_currentThreadId); } else ERROR_LOG(m_logger) << "Could not get PrtCnvDbThread!"; } QTimer::singleShot(m_calcTimeout, this, &PrtCnvRadioTrafficThread::onCalculateTrafficLoad); } void PrtCnvRadioTrafficThread::onCleanupTimeout() { const time_t curTime = time(nullptr); { QMutexLocker l(&m_mutex); QMap::iterator it = m_data.begin(); while (it != m_data.end()) { const time_t key = it.key(); if (key < curTime - m_cleanupTimeout - 10) { m_data.erase(it++); m_telegrams.remove(key); continue; } break; } } QTimer::singleShot(m_cleanupTimeout, this, &PrtCnvRadioTrafficThread::onCleanupTimeout); } }