ETM-Schurig/powersync-core
Archived
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge PR #600: NetworkDeviceDiscovery: wait for pending MAC address manufacturer lookups before finishing a discovery

Browse Source
This commit is contained in:
jenkins 2022-12-13 21:04:59 +01:00
commit 66c606aa10
4 changed files with 130 additions and 65 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

135
libnymea-core/hardware/network/networkdevicediscoveryimpl.cpp
View File

@ -64,7 +64,21 @@ NetworkDeviceDiscoveryImpl::NetworkDeviceDiscoveryImpl(QObject *parent) :
m_discoveryTimer->setInterval(20000); m_discoveryTimer->setInterval(20000);
m_discoveryTimer->setSingleShot(true); m_discoveryTimer->setSingleShot(true);
connect(m_discoveryTimer, &QTimer::timeout, this, [=](){ connect(m_discoveryTimer, &QTimer::timeout, this, [=](){
if (m_runningPingReplies.isEmpty() && m_currentReply) { if (!m_runningPingReplies.isEmpty()) {
qCDebug(dcNetworkDeviceDiscovery()) << "Discovery timeout occurred. There are still" << m_runningPingReplies.count() << "ping replies pending and" << m_ping->queueCount() << "addresses int the ping queue. Aborting them...";
foreach (PingReply *reply, m_runningPingReplies) {
reply->abort();
}
}
// We still wait for any mac manufacturer lookups, since we got already a mac...
if (!m_runningMacDatabaseReplies.isEmpty()) {
qCDebug(dcNetworkDeviceDiscovery()) << "Discovery timeout occurred but there are still" << m_runningMacDatabaseReplies.count() << "mac database replies pending. Waiting for them to finish...";
return;
}
if (m_currentDiscoveryReply) {
qCDebug(dcNetworkDeviceDiscovery()) << "Discovery timeout occurred and all pending replies have finished.";
finishDiscovery(); finishDiscovery();
} }
}); });
@ -103,27 +117,27 @@ NetworkDeviceDiscoveryReply *NetworkDeviceDiscoveryImpl::discover()
// reply and owns the object, even if the discovery has been finished. // reply and owns the object, even if the discovery has been finished.
// Create the internal object if required // Create the internal object if required
bool alreadyRunning = (m_currentReply != nullptr); bool alreadyRunning = (m_currentDiscoveryReply != nullptr);
if (alreadyRunning) { if (alreadyRunning) {
if (m_currentReply->isFinished()) { if (m_currentDiscoveryReply->isFinished()) {
qCDebug(dcNetworkDeviceDiscovery()) << "Discovery internally already running and finished."; qCDebug(dcNetworkDeviceDiscovery()) << "Discovery internally already running and finished.";
} else { } else {
qCDebug(dcNetworkDeviceDiscovery()) << "Discovery internally already running. Re-using the current running discovery reply."; qCDebug(dcNetworkDeviceDiscovery()) << "Discovery internally already running. Re-using the current running discovery reply.";
} }
} else { } else {
qCDebug(dcNetworkDeviceDiscovery()) << "Starting internally a new discovery."; qCDebug(dcNetworkDeviceDiscovery()) << "Starting internally a new discovery.";
m_currentReply = new NetworkDeviceDiscoveryReplyImpl(this); m_currentDiscoveryReply = new NetworkDeviceDiscoveryReplyImpl(this);
connect(m_currentReply, &NetworkDeviceDiscoveryReplyImpl::networkDeviceInfoAdded, this, &NetworkDeviceDiscoveryImpl::updateCache); connect(m_currentDiscoveryReply, &NetworkDeviceDiscoveryReplyImpl::networkDeviceInfoAdded, this, &NetworkDeviceDiscoveryImpl::updateCache);
connect(m_currentReply, &NetworkDeviceDiscoveryReplyImpl::finished, this, [this](){ connect(m_currentDiscoveryReply, &NetworkDeviceDiscoveryReplyImpl::finished, this, [this](){
// Finish all pending replies // Finish all pending replies
foreach (NetworkDeviceDiscoveryReplyImpl *reply, m_pendingReplies) { foreach (NetworkDeviceDiscoveryReplyImpl *reply, m_pendingDiscoveryReplies) {
// Sync all network device infos with all pending replies // Sync all network device infos with all pending replies
foreach (const NetworkDeviceInfo &info, m_currentReply->networkDeviceInfos()) { foreach (const NetworkDeviceInfo &info, m_currentDiscoveryReply->networkDeviceInfos()) {
reply->addCompleteNetworkDeviceInfo(info); reply->addCompleteNetworkDeviceInfo(info);
} }
foreach (const NetworkDeviceInfo &info, m_currentReply->virtualNetworkDeviceInfos()) { foreach (const NetworkDeviceInfo &info, m_currentDiscoveryReply->virtualNetworkDeviceInfos()) {
reply->addVirtualNetworkDeviceInfo(info); reply->addVirtualNetworkDeviceInfo(info);
} }
} }
@ -131,11 +145,11 @@ NetworkDeviceDiscoveryReply *NetworkDeviceDiscoveryImpl::discover()
// Delete the current reply before finishing the pending replies. // Delete the current reply before finishing the pending replies.
// Just in case some one restarts a discovery on finished, a new internal // Just in case some one restarts a discovery on finished, a new internal
// object should be created // object should be created
m_currentReply->deleteLater(); m_currentDiscoveryReply->deleteLater();
m_currentReply = nullptr; m_currentDiscoveryReply = nullptr;
foreach (NetworkDeviceDiscoveryReplyImpl *reply, m_pendingReplies) { foreach (NetworkDeviceDiscoveryReplyImpl *reply, m_pendingDiscoveryReplies) {
m_pendingReplies.removeAll(reply); m_pendingDiscoveryReplies.removeAll(reply);
reply->setFinished(true); reply->setFinished(true);
emit reply->finished(); emit reply->finished();
} }
@ -144,9 +158,9 @@ NetworkDeviceDiscoveryReply *NetworkDeviceDiscoveryImpl::discover()
// Create the reply for the user // Create the reply for the user
NetworkDeviceDiscoveryReplyImpl *reply = new NetworkDeviceDiscoveryReplyImpl(this); NetworkDeviceDiscoveryReplyImpl *reply = new NetworkDeviceDiscoveryReplyImpl(this);
connect(m_currentReply, &NetworkDeviceDiscoveryReplyImpl::networkDeviceInfoAdded, reply, &NetworkDeviceDiscoveryReplyImpl::addCompleteNetworkDeviceInfo); connect(m_currentDiscoveryReply, &NetworkDeviceDiscoveryReplyImpl::networkDeviceInfoAdded, reply, &NetworkDeviceDiscoveryReplyImpl::addCompleteNetworkDeviceInfo);
connect(m_currentReply, &NetworkDeviceDiscoveryReplyImpl::hostAddressDiscovered, reply, &NetworkDeviceDiscoveryReplyImpl::hostAddressDiscovered); connect(m_currentDiscoveryReply, &NetworkDeviceDiscoveryReplyImpl::hostAddressDiscovered, reply, &NetworkDeviceDiscoveryReplyImpl::hostAddressDiscovered);
m_pendingReplies.append(reply); m_pendingDiscoveryReplies.append(reply);
if (!available()) { if (!available()) {
qCWarning(dcNetworkDeviceDiscovery()) << "The network discovery is not available. Please make sure the binary has the required capability (CAP_NET_RAW) or start the application as root."; qCWarning(dcNetworkDeviceDiscovery()) << "The network discovery is not available. Please make sure the binary has the required capability (CAP_NET_RAW) or start the application as root.";
@ -159,10 +173,10 @@ NetworkDeviceDiscoveryReply *NetworkDeviceDiscoveryImpl::discover()
// Add already discovered network device infos in the next event loop // Add already discovered network device infos in the next event loop
// so any connections after this method call will work as expected // so any connections after this method call will work as expected
QTimer::singleShot(0, reply, [this, reply](){ QTimer::singleShot(0, reply, [this, reply](){
if (!m_currentReply) if (!m_currentDiscoveryReply)
return; return;
foreach (const NetworkDeviceInfo &networkDeviceInfo, m_currentReply->networkDeviceInfos()) { foreach (const NetworkDeviceInfo &networkDeviceInfo, m_currentDiscoveryReply->networkDeviceInfos()) {
reply->addCompleteNetworkDeviceInfo(networkDeviceInfo); reply->addCompleteNetworkDeviceInfo(networkDeviceInfo);
} }
}); });
@ -372,13 +386,18 @@ void NetworkDeviceDiscoveryImpl::pingAllNetworkDevices()
m_runningPingReplies.removeAll(reply); m_runningPingReplies.removeAll(reply);
if (reply->error() == PingReply::ErrorNoError) { if (reply->error() == PingReply::ErrorNoError) {
qCDebug(dcNetworkDeviceDiscovery()) << "Ping response from" << targetAddress.toString() << reply->hostName() << reply->duration() << "ms"; qCDebug(dcNetworkDeviceDiscovery()) << "Ping response from" << targetAddress.toString() << reply->hostName() << reply->duration() << "ms";
if (m_currentReply) { if (m_currentDiscoveryReply) {
m_currentReply->processPingResponse(targetAddress, reply->hostName()); m_currentDiscoveryReply->processPingResponse(targetAddress, reply->hostName());
} }
} }
if (m_runningPingReplies.isEmpty() && m_currentReply && !m_discoveryTimer->isActive()) { if (m_runningPingReplies.isEmpty() && !m_runningMacDatabaseReplies.isEmpty()) {
qCWarning(dcNetworkDeviceDiscovery()) << "All ping replies finished for discovery." << m_currentReply->networkDeviceInfos().count(); qCDebug(dcNetworkDeviceDiscovery()) << "All ping replies have finished but there are still" << m_runningMacDatabaseReplies.count() << "mac db lookups pending. Waiting for them to finish...";
return;
}
if (m_runningPingReplies.isEmpty() && m_runningMacDatabaseReplies.isEmpty() && m_currentDiscoveryReply && !m_discoveryTimer->isActive()) {
qCDebug(dcNetworkDeviceDiscovery()) << "All pending replies have finished.";
finishDiscovery(); finishDiscovery();
} }
}); });
@ -592,39 +611,51 @@ void NetworkDeviceDiscoveryImpl::processArpTraffic(const QNetworkInterface &inte
} }
} }
// Check if we have a reply running // Check if we have currently reply running
if (m_currentReply) { if (!m_currentDiscoveryReply)
return;
// First process the response // First process the response
m_currentReply->processArpResponse(interface, address, macAddress); m_currentDiscoveryReply->processArpResponse(interface, address, macAddress);
// Check if we know the mac address manufacturer from the cache // Check if we know the mac address manufacturer from the cache
bool requiresMacAddressLookup = true; bool requiresMacAddressLookup = true;
if (m_networkInfoCache.contains(macAddress)) { if (m_networkInfoCache.contains(macAddress)) {
QString cachedManufacturer = m_networkInfoCache[macAddress].macAddressManufacturer(); QString cachedManufacturer = m_networkInfoCache[macAddress].macAddressManufacturer();
if (!cachedManufacturer.isEmpty()) { if (!cachedManufacturer.isEmpty()) {
// Found the mac address manufacturer in the cache, let's use that one... // Found the mac address manufacturer in the cache, let's use that one...
m_currentReply->processMacManufacturer(macAddress, cachedManufacturer); qCDebug(dcNetworkDeviceDiscovery()) << "Using cached manufacturer " << cachedManufacturer << "for" << macAddress.toString();
requiresMacAddressLookup = false; m_currentDiscoveryReply->processMacManufacturer(macAddress, cachedManufacturer);
} requiresMacAddressLookup = false;
} }
}
if (requiresMacAddressLookup) { if (!requiresMacAddressLookup)
// Lookup the mac address vendor if possible return;
if (m_macAddressDatabase->available()) {
// Not found in the cache, and the mac address database is available...let's make a query // Lookup the mac address vendor if possible
MacAddressDatabaseReply *reply = m_macAddressDatabase->lookupMacAddress(macAddress.toString()); if (m_macAddressDatabase->available()) {
connect(reply, &MacAddressDatabaseReply::finished, m_currentReply, [=](){ // Not found in the cache, and the mac address database is available...let's make a query
// Note: set the mac manufacturer explicitly to make the info complete (even an empty sring) MacAddressDatabaseReply *reply = m_macAddressDatabase->lookupMacAddress(macAddress.toString());
qCDebug(dcNetworkDeviceDiscovery()) << "MAC manufacturer lookup finished for" << macAddress << ":" << reply->manufacturer(); m_runningMacDatabaseReplies.append(reply);
m_currentReply->processMacManufacturer(macAddress, reply->manufacturer()); connect(reply, &MacAddressDatabaseReply::finished, this, [this, macAddress, reply](){
}); m_runningMacDatabaseReplies.removeAll(reply);
} else {
// Not found in the cache, and no mac address database available...we are done with mac vendor // Note: set the mac manufacturer explicitly to make the info complete (even an empty sring)
// Note: set the mac manufacturer explicitly to make the info complete qCDebug(dcNetworkDeviceDiscovery()) << "MAC manufacturer lookup finished for" << macAddress << ":" << reply->manufacturer();
m_currentReply->processMacManufacturer(macAddress, QString()); if (m_currentDiscoveryReply) {
m_currentDiscoveryReply->processMacManufacturer(macAddress, reply->manufacturer());
if (m_runningPingReplies.isEmpty() && m_runningMacDatabaseReplies.isEmpty() && !m_discoveryTimer->isActive()) {
qCWarning(dcNetworkDeviceDiscovery()) << "All pending replies have finished.";
finishDiscovery();
}
} }
} });
} else {
// Not found in the cache, and no mac address database available...we are done with mac vendor
// Set the mac manufacturer explicitly to make the info complete
m_currentDiscoveryReply->processMacManufacturer(macAddress, QString());
} }
} }
@ -722,8 +753,8 @@ void NetworkDeviceDiscoveryImpl::finishDiscovery()
m_lastDiscovery = QDateTime::currentDateTime(); m_lastDiscovery = QDateTime::currentDateTime();
// Clean up internal reply // Clean up internal reply
if (m_currentReply) { if (m_currentDiscoveryReply) {
m_currentReply->processDiscoveryFinished(); m_currentDiscoveryReply->processDiscoveryFinished();
} }
} }

5
libnymea-core/hardware/network/networkdevicediscoveryimpl.h
View File

@ -101,8 +101,9 @@ private:
uint m_monitorInterval = 60; // 1 min uint m_monitorInterval = 60; // 1 min
uint m_cacheCleanupPeriod = 30; // days uint m_cacheCleanupPeriod = 30; // days
NetworkDeviceDiscoveryReplyImpl *m_currentReply = nullptr; NetworkDeviceDiscoveryReplyImpl *m_currentDiscoveryReply = nullptr;
QList<NetworkDeviceDiscoveryReplyImpl *> m_pendingReplies; QList<NetworkDeviceDiscoveryReplyImpl *> m_pendingDiscoveryReplies;
QList<MacAddressDatabaseReply *> m_runningMacDatabaseReplies;
QList<PingReply *> m_runningPingReplies; QList<PingReply *> m_runningPingReplies;
QHash<MacAddress, NetworkDeviceMonitorImpl *> m_monitors; QHash<MacAddress, NetworkDeviceMonitorImpl *> m_monitors;

51
libnymea/network/ping.cpp
View File

@ -50,7 +50,7 @@ NYMEA_LOGGING_CATEGORY(dcPingTraffic, "PingTraffic")
Ping::Ping(QObject *parent) : QObject(parent) Ping::Ping(QObject *parent) : QObject(parent)
{ {
m_queueTimer = new QTimer(this); m_queueTimer = new QTimer(this);
m_queueTimer->setInterval(20); m_queueTimer->setInterval(10);
m_queueTimer->setSingleShot(true); m_queueTimer->setSingleShot(true);
connect(m_queueTimer, &QTimer::timeout, this, [=](){ connect(m_queueTimer, &QTimer::timeout, this, [=](){
sendNextReply(); sendNextReply();
@ -111,6 +111,11 @@ PingReply::Error Ping::error() const
return m_error; return m_error;
} }
int Ping::queueCount() const
{
return m_replyQueue.count();
}
PingReply *Ping::ping(const QHostAddress &hostAddress, uint retries) PingReply *Ping::ping(const QHostAddress &hostAddress, uint retries)
{ {
PingReply *reply = createReply(hostAddress); PingReply *reply = createReply(hostAddress);
@ -144,10 +149,15 @@ void Ping::sendNextReply()
if (m_replyQueue.isEmpty()) if (m_replyQueue.isEmpty())
return; return;
PingReply *reply = m_replyQueue.dequeue(); m_currentReply = m_replyQueue.dequeue();
qCDebug(dcPing()) << "Send next reply," << m_replyQueue.count() << "left in queue"; qCDebug(dcPing()) << "Send next reply " << m_currentReply->targetHostAddress().toString() << QString("0x%1").arg(m_currentReply->requestId(), 4, 16, QChar('0')) << ", " << m_replyQueue.count() << "left in queue";
m_queueTimer->start(); m_queueTimer->start();
QTimer::singleShot(0, reply, [=]() { performPing(reply); }); QTimer::singleShot(0, this, [=]() {
if (!m_currentReply)
return;
performPing(m_currentReply);
});
} }
void Ping::performPing(PingReply *reply) void Ping::performPing(PingReply *reply)
@ -298,7 +308,7 @@ PingReply *Ping::createReply(const QHostAddress &hostAddress)
reply->m_networkInterface = NetworkUtils::getInterfaceForHostaddress(hostAddress); reply->m_networkInterface = NetworkUtils::getInterfaceForHostaddress(hostAddress);
connect(reply, &PingReply::timeout, this, [=](){ connect(reply, &PingReply::timeout, this, [=](){
// Note: this is not the ICMP timeout, here we actually got nothing from nobody... // Note: this is not the ICMP timeout, here we actually got nothing from anybody...
finishReply(reply, PingReply::ErrorTimeout); finishReply(reply, PingReply::ErrorTimeout);
}); });
@ -307,11 +317,13 @@ PingReply *Ping::createReply(const QHostAddress &hostAddress)
}); });
connect(reply, &PingReply::finished, this, [=](){ connect(reply, &PingReply::finished, this, [=](){
cleanUpReply(reply);
reply->deleteLater(); reply->deleteLater();
});
// Cleanup any retry left over queue stuff // Just in case the reply get's deleted before beeing able to finish ...
m_pendingReplies.remove(reply->requestId()); connect(reply, &PingReply::destroyed, this, [=](){
m_replyQueue.removeAll(reply); cleanUpReply(reply);
}); });
return reply; return reply;
@ -353,6 +365,23 @@ void Ping::finishReply(PingReply *reply, PingReply::Error error)
} }
} }
void Ping::cleanUpReply(PingReply *reply)
{
// Cleanup any retry left over queue stuff
if (m_currentReply == reply)
m_currentReply = nullptr;
m_pendingReplies.remove(reply->requestId());
m_replyQueue.removeAll(reply);
if (m_pendingHostLookups.values().contains(reply)) {
// Abort any pending host lookups, the reply has been finished
int lookupId = m_pendingHostLookups.key(reply);
QHostInfo::abortHostLookup(lookupId);
m_pendingHostLookups.remove(lookupId);
}
}
void Ping::onSocketReadyRead(int socketDescriptor) void Ping::onSocketReadyRead(int socketDescriptor)
{ {
// We must read all data otherwise the socket notifier does not work as expected // We must read all data otherwise the socket notifier does not work as expected
@ -391,7 +420,7 @@ void Ping::onSocketReadyRead(int socketDescriptor)
if (responsePacket->icmp_type == ICMP_ECHOREPLY) { if (responsePacket->icmp_type == ICMP_ECHOREPLY) {
PingReply *reply = m_pendingReplies.take(icmpId); PingReply *reply = m_pendingReplies.value(icmpId);
if (!reply) { if (!reply) {
qCDebug(dcPing()) << "No pending reply for ping echo response with id" << QString("0x%1").arg(icmpId, 4, 16, QChar('0')) << "Sequence:" << icmpSequnceNumber << "from" << senderAddress.toString(); qCDebug(dcPing()) << "No pending reply for ping echo response with id" << QString("0x%1").arg(icmpId, 4, 16, QChar('0')) << "Sequence:" << icmpSequnceNumber << "from" << senderAddress.toString();
return; return;
@ -426,12 +455,12 @@ void Ping::onSocketReadyRead(int socketDescriptor)
// Note: due to a Qt bug < 5.9 we need to use old SLOT style and cannot make use of lambda here // Note: due to a Qt bug < 5.9 we need to use old SLOT style and cannot make use of lambda here
int lookupId = QHostInfo::lookupHost(senderAddress.toString(), this, SLOT(onHostLookupFinished(QHostInfo))); int lookupId = QHostInfo::lookupHost(senderAddress.toString(), this, SLOT(onHostLookupFinished(QHostInfo)));
m_pendingHostLookups.insert(lookupId, reply); m_pendingHostLookups.insert(lookupId, reply);
// Finish the reply after the host lookup has finished
} else { } else {
finishReply(reply, PingReply::ErrorNoError); finishReply(reply, PingReply::ErrorNoError);
} }
} else if (responsePacket->icmp_type == ICMP_DEST_UNREACH) { } else if (responsePacket->icmp_type == ICMP_DEST_UNREACH) {
// Get the sending package // Get the sending package
@ -463,7 +492,7 @@ void Ping::onSocketReadyRead(int socketDescriptor)
<< "ID:" << QString("0x%1").arg(icmpId, 4, 16, QChar('0')) << "ID:" << QString("0x%1").arg(icmpId, 4, 16, QChar('0'))
<< "Sequence:" << icmpSequnceNumber; << "Sequence:" << icmpSequnceNumber;
PingReply *reply = m_pendingReplies.take(icmpId); PingReply *reply = m_pendingReplies.value(icmpId);
if (!reply) { if (!reply) {
qCDebug(dcPingTraffic()) << "No pending reply for ping echo response unreachable with ID" qCDebug(dcPingTraffic()) << "No pending reply for ping echo response unreachable with ID"
<< QString("0x%1").arg(icmpId, 4, 16, QChar('0')) << QString("0x%1").arg(icmpId, 4, 16, QChar('0'))

4
libnymea/network/ping.h
View File

@ -62,6 +62,8 @@ public:
PingReply::Error error() const; PingReply::Error error() const;
int queueCount() const;
PingReply *ping(const QHostAddress &hostAddress, uint retries = 3); PingReply *ping(const QHostAddress &hostAddress, uint retries = 3);
PingReply *ping(const QHostAddress &hostAddress, bool lookupHost, uint retries = 3); PingReply *ping(const QHostAddress &hostAddress, bool lookupHost, uint retries = 3);
@ -87,6 +89,7 @@ private:
QQueue<PingReply *> m_replyQueue; QQueue<PingReply *> m_replyQueue;
QTimer *m_queueTimer = nullptr; QTimer *m_queueTimer = nullptr;
PingReply *m_currentReply = nullptr;
void sendNextReply(); void sendNextReply();
QHash<int, PingReply *> m_pendingHostLookups; QHash<int, PingReply *> m_pendingHostLookups;
@ -102,6 +105,7 @@ private:
PingReply *createReply(const QHostAddress &hostAddress); PingReply *createReply(const QHostAddress &hostAddress);
void finishReply(PingReply *reply, PingReply::Error error); void finishReply(PingReply *reply, PingReply::Error error);
void cleanUpReply(PingReply *reply);
private slots: private slots:
void onSocketReadyRead(int socketDescriptor); void onSocketReadyRead(int socketDescriptor);