upnp: add IGD discovery phase.

When the protocols start searching for IGD, the IGD discovery phase begins.
During this phase, all requested mappings will be marked as pending.
If an IGD is found, the pending mappings will be requested from the router.
If the phase ends due to a timeout, the pending mappings will be marked as failed

GitLab: #36
Change-Id: Icb19b05e40968d619cb032a949a8fe8acfef311b

Author: AmnaSnene

include/upnp/upnp_context.h

@@ -75,6 +75,7 @@ class UpnpMappingObserver
     virtual void onMappingRequestFailed(const Mapping& map) = 0;
     virtual void onMappingRenewed(const std::shared_ptr<IGD>& igd, const Mapping& map) = 0;
     virtual void onMappingRemoved(const std::shared_ptr<IGD>& igd, const Mapping& map) = 0;
+    virtual void onIgdDiscoveryStarted() = 0;
 };
 
 class UPnPContext : public UpnpMappingObserver
@@ -133,6 +134,11 @@ class UPnPContext : public UpnpMappingObserver
         });
     }
 
+    // Set the timeout for the IGD discovery process.
+    // If the timeout expires and no valid IGD has been discovered,
+    // then the state of all pending mappings is set to FAILED.
+    void setIgdDiscoveryTimeout(std::chrono::milliseconds timeout);
+
 private:
     // Initialization
     void init();
@@ -242,6 +248,9 @@ class UPnPContext : public UpnpMappingObserver
     // Callback used to report remove request status.
     void onMappingRemoved(const std::shared_ptr<IGD>& igd, const Mapping& map) override;
 
+    // Callback used to report the start of the discovery process: search for IGDs.
+    void onIgdDiscoveryStarted() override;
+
 private:
     UPnPContext(const UPnPContext&) = delete;
     UPnPContext(UPnPContext&&) = delete;
@@ -304,6 +313,16 @@ class UPnPContext : public UpnpMappingObserver
     // Shutdown synchronization
     bool shutdownComplete_ {false};
     bool shutdownTimedOut_ {false};
+
+    // IGD Discovery synchronization. This boolean indicates if the IGD discovery is in progress.
+    bool igdDiscoveryInProgress_ {true};
+    std::mutex igdDiscoveryMutex_;
+    std::chrono::milliseconds igdDiscoveryTimeout_ {std::chrono::milliseconds(500)};
+
+    // End of the discovery process.
+    void _endIgdDiscovery();
+
+    asio::steady_timer igdDiscoveryTimer_;
 };
 
 } // namespace upnp

src/upnp/protocol/natpmp/nat_pmp.cpp

@@ -189,14 +189,14 @@ void
 NatPmp::searchForIgd()
 {
     if (not initialized_) {
+        observer_->onIgdDiscoveryStarted();
         initNatPmp();
     }
 
     // Schedule a retry in case init failed.
     if (not initialized_) {
         if (igdSearchCounter_++ < MAX_RESTART_SEARCH_RETRIES) {
             if (logger_) logger_->debug("NAT-PMP: Start search for IGDs. Attempt {}", igdSearchCounter_);
-
             // Cancel the current timer (if any) and re-schedule.
             searchForIgdTimer_.expires_after(NATPMP_SEARCH_RETRY_UNIT * igdSearchCounter_);
             searchForIgdTimer_.async_wait([w=weak()](const asio::error_code& ec) {

src/upnp/protocol/pupnp/pupnp.cpp

@@ -240,41 +240,46 @@ PUPnP::terminate()
     if (logger_) logger_->debug("PUPnP: Instance {} terminated", fmt::ptr(this));
 }
 
+void
+PUPnP::searchForDeviceAsync(const std::string& deviceType)
+{
+    // Despite its name and the claim in the libupnp documentation that it "returns immediately",
+    // the UpnpSearchAsync function isn't really async. This is because it tries to send multiple
+    // copies of each search message and waits for a certain amount of time after sending each
+    // copy. The number of copies is given by the NUM_SSDP_COPY macro, whose default value is 2,
+    // and the waiting time is determined by the SSDP_PAUSE macro, whose default value is 100 (ms).
+    // If both IPv4 and IPv6 are enabled, then UpnpSearchAsync sends 3 distinct messages (2 for IPv6
+    // and 1 for IPv4), resulting in a total of 3 * 2 * 100 = 600 ms spent waiting by default.
+    // This is why we put the call to UpnpSearchAsync on its own thread.
+    dht::ThreadPool::io().run([w = weak_from_this(), deviceType] {
+        auto sthis = std::static_pointer_cast<PUPnP>(w.lock());
+        if (!sthis)
+            return;
+
+        auto err = UpnpSearchAsync(sthis->ctrlptHandle_,
+                                   SEARCH_TIMEOUT,
+                                   deviceType.c_str(),
+                                   sthis.get());
+        if (err != UPNP_E_SUCCESS) {
+            if (sthis->logger_)
+                sthis->logger_->warn("PUPnP: Send search for {} failed. Error {:d}: {}",
+                                     deviceType,
+                                     err,
+                                     UpnpGetErrorMessage(err));
+        }
+    });
+}
 void
 PUPnP::searchForDevices()
 {
     if (logger_) logger_->debug("PUPnP: Send IGD search request");
 
     // Send out search for multiple types of devices, as some routers may possibly
     // only reply to one.
-
-    auto err = UpnpSearchAsync(ctrlptHandle_, SEARCH_TIMEOUT, UPNP_ROOT_DEVICE, this);
-    if (err != UPNP_E_SUCCESS) {
-        if (logger_) logger_->warn("PUPnP: Send search for UPNP_ROOT_DEVICE failed. Error {:d}: {}",
-                  err,
-                  UpnpGetErrorMessage(err));
-    }
-
-    err = UpnpSearchAsync(ctrlptHandle_, SEARCH_TIMEOUT, UPNP_IGD_DEVICE, this);
-    if (err != UPNP_E_SUCCESS) {
-        if (logger_) logger_->warn("PUPnP: Send search for UPNP_IGD_DEVICE failed. Error {:d}: {}",
-                  err,
-                  UpnpGetErrorMessage(err));
-    }
-
-    err = UpnpSearchAsync(ctrlptHandle_, SEARCH_TIMEOUT, UPNP_WANIP_SERVICE, this);
-    if (err != UPNP_E_SUCCESS) {
-        if (logger_) logger_->warn("PUPnP: Send search for UPNP_WANIP_SERVICE failed. Error {:d}: {}",
-                  err,
-                  UpnpGetErrorMessage(err));
-    }
-
-    err = UpnpSearchAsync(ctrlptHandle_, SEARCH_TIMEOUT, UPNP_WANPPP_SERVICE, this);
-    if (err != UPNP_E_SUCCESS) {
-        if (logger_) logger_->warn("PUPnP: Send search for UPNP_WANPPP_SERVICE failed. Error {:d}: {}",
-                  err,
-                  UpnpGetErrorMessage(err));
-    }
+    searchForDeviceAsync(UPNP_ROOT_DEVICE);
+    searchForDeviceAsync(UPNP_IGD_DEVICE);
+    searchForDeviceAsync(UPNP_WANIP_SERVICE);
+    searchForDeviceAsync(UPNP_WANPPP_SERVICE);
 }
 
 void
@@ -340,6 +345,7 @@ PUPnP::searchForIgd()
         if (clientRegistered_) {
             assert(initialized_);
             searchForDevices();
+            observer_->onIgdDiscoveryStarted();
         } else {
             if (logger_) logger_->warn("PUPnP: PUPNP not fully setup. Skipping the IGD search");
         }

src/upnp/protocol/pupnp/pupnp.h

@@ -134,6 +134,9 @@ class PUPnP : public UPnPProtocol
     // Start search for UPNP devices
     void searchForDevices();
 
+    // Start search for UPNP device in a different thread
+    void searchForDeviceAsync(const std::string& deviceType);
+
     // Return true if it has at least one valid IGD.
     bool hasValidIgd() const;
 

src/upnp/upnp_context.cpp

@@ -51,6 +51,8 @@ UPnPContext::UPnPContext(const std::shared_ptr<asio::io_context>& ioContext, con
  , renewalSchedulingTimer_(*ctx)
  , syncTimer_(*ctx)
  , connectivityChangedTimer_(*ctx)
+ , igdDiscoveryTimer_(*ctx)
+
 {
     if (logger_) logger_->debug("Creating UPnPContext instance [{}]", fmt::ptr(this));
 
@@ -1095,6 +1097,51 @@ UPnPContext::onMappingRemoved(const std::shared_ptr<IGD>& igd, const Mapping& ma
         map->getNotifyCallback()(map);
 }
 
+void
+UPnPContext::onIgdDiscoveryStarted()
+{
+    std::lock_guard lock(igdDiscoveryMutex_);
+    igdDiscoveryInProgress_ = true;
+    if (logger_) logger_->debug("IGD Discovery started");
+    igdDiscoveryTimer_.expires_after(igdDiscoveryTimeout_);
+    igdDiscoveryTimer_.async_wait([this] (const asio::error_code& ec) {
+        if (ec != asio::error::operation_aborted && igdDiscoveryInProgress_) {
+            _endIgdDiscovery();
+        }
+    });
+}
+
+void
+UPnPContext::_endIgdDiscovery()
+{
+    std::lock_guard lockDiscovery_(igdDiscoveryMutex_);
+    igdDiscoveryInProgress_ = false;
+    if (logger_) logger_->debug("IGD Discovery ended");
+    if (isReady()) {
+       return;
+    }
+    // if there is no valid IGD, the pending mapping requests will be changed to failed
+    std::lock_guard lockMappings_(mappingMutex_);
+    PortType types[2] {PortType::TCP, PortType::UDP};
+    for (auto& type : types) {
+        const auto& mappingList = getMappingList(type);
+        for (auto const& [_, map] : mappingList) {
+            updateMappingState(map, MappingState::FAILED);
+            // Do not unregister the mapping, it's up to the controller to decide. It will be unregistered when the controller releases it.
+            // unregisterMapping(map) here will cause a deadlock because of the lock on mappingMutex_.
+            if (logger_) logger_->warn("Request for mapping {} failed, no IGD available",
+                                       map->toString());
+        }
+    }
+}
+
+void
+UPnPContext::setIgdDiscoveryTimeout(std::chrono::milliseconds timeout)
+{
+    std::lock_guard lock(igdDiscoveryMutex_);
+    igdDiscoveryTimeout_ = timeout;
+}
+
 Mapping::sharedPtr_t
 UPnPContext::registerMapping(Mapping& map)
 {
@@ -1123,14 +1170,23 @@ UPnPContext::registerMapping(Mapping& map)
         assert(mapPtr);
     }
 
-    // No available IGD. The pending mapping requests will be processed
-    // when an IGD becomes available
     if (not isReady()) {
-        if (logger_) logger_->warn("No IGD available. Mapping will be requested when an IGD becomes available");
+        // There is no valid IGD available
+        std::lock_guard lock(igdDiscoveryMutex_);
+        // IGD discovery is in progress, the mapping request will be made once an IGD becomes available
+        if (igdDiscoveryInProgress_) {
+            if (logger_) logger_->debug("Request for mapping {} will be requested when an IGD becomes available",
+                                        map.toString());
+        } else {
+            // it's not in the IGD discovery phase, the mapping request will fail
+            if (logger_) logger_->warn("Request for mapping {} failed, no IGD available",
+                                       map.toString());
+            updateMappingState(mapPtr, MappingState::FAILED);
+        }
     } else {
+        // There is a valid IGD available, request the mapping.
         requestMapping(mapPtr);
     }
-
     return mapPtr;
 }