Fix for large values of policy unresolved counts

Whenever a connection is destroyed, we need to clear the pending resolutions for that connection. Without this, unresolved counts and pending resolutions can get out of sync. Signed-off-by: Gautam Venkataramanan <gautam.chennai@gmail.com>
noironetworks · Jun 29, 2020 · bfc1098 · bfc1098
1 parent f0854f9
commit bfc1098
Show file tree

Hide file tree

Showing 4 changed files with 30 additions and 5 deletions.
diff --git a/libopflex/engine/OpflexClientConnection.cpp b/libopflex/engine/OpflexClientConnection.cpp
@@ -45,6 +45,7 @@ OpflexClientConnection::OpflexClientConnection(HandlerFactory& handlerFactory,
 }
 
 OpflexClientConnection::~OpflexClientConnection() {
+    pool->clearPendingItems(this);
 }
 
 const string& OpflexClientConnection::getName() {

diff --git a/libopflex/engine/OpflexPool.cpp b/libopflex/engine/OpflexPool.cpp
@@ -46,20 +46,39 @@ OpflexPool::~OpflexPool() {
 }
 
 void OpflexPool::addPendingItem(OpflexClientConnection* conn, const std::string& uri) {
-    std::string hostName = conn->getHostname();
+    const std::string& hostName = conn->getHostname();
     std::unique_lock<std::mutex> lock(modify_uri_mutex);
-    if(pendingResolution[hostName].insert(uri).second == true) {
-       conn->getOpflexStats()->incrPolUnresolvedCount();
+    if (pendingResolution[hostName].insert(uri).second == true) {
+        conn->getOpflexStats()->incrPolUnresolvedCount();
+        LOG(TRACE) << "add pending UC: conn: " << hostName
+                   << " size: " << pendingResolution[hostName].size()
+                   << " UC: " << conn->getOpflexStats()->getPolUnresolvedCount()
+                   << " URI: " << uri;
+    }
+}
+
+void OpflexPool::clearPendingItems (OpflexClientConnection* conn)
+{
+    const std::string& hostName = conn->getHostname();
+    std::unique_lock<std::mutex> lock(modify_uri_mutex);
+    if (pendingResolution.find(hostName) != pendingResolution.end()) {
+        pendingResolution[hostName].clear();
+        pendingResolution.erase(hostName);
+        LOG(TRACE) << "clearing pending UCs: conn: " << hostName;
     }
 }
 
 void OpflexPool::removePendingItem(OpflexClientConnection* conn, const std::string& uri) {
-    std::string hostName = conn->getHostname();
+    const std::string& hostName = conn->getHostname();
     std::unique_lock<std::mutex> lock(modify_uri_mutex);
     std::set<std::string>::iterator rem = pendingResolution[hostName].find(uri);
     if (rem != pendingResolution[hostName].end()) {
         pendingResolution[hostName].erase(rem);
         conn->getOpflexStats()->decrPolUnresolvedCount();
+        LOG(TRACE) << "del pending UC: conn: " << hostName
+                   << " size: " << pendingResolution[hostName].size()
+                   << " UC: " << conn->getOpflexStats()->getPolUnresolvedCount()
+                   << " URI: " << uri;
     }
 }
 

diff --git a/libopflex/engine/include/opflex/engine/Processor.h b/libopflex/engine/include/opflex/engine/Processor.h
@@ -308,7 +308,7 @@ class Processor : public internal::AbstractObjectListener,
         NEW,
         /** a local item that's been updated */
         UPDATED,
-        /** a local item that's been send to the server */
+        /** a local item that's been sent to the server */
         IN_SYNC,
         /** A remote item that does not get resolved */
         REMOTE,

diff --git a/libopflex/engine/include/opflex/engine/internal/OpflexPool.h b/libopflex/engine/include/opflex/engine/internal/OpflexPool.h
@@ -215,6 +215,11 @@ class OpflexPool : private boost::noncopyable {
      */
     void removePendingItem(OpflexClientConnection* conn, const std::string& uri);
 
+    /**
+     * Clear the pending policies per connection
+     */
+    void clearPendingItems(OpflexClientConnection* conn);
+
     /**
      * Register the given peer status listener to get updates on the
      * health of the connection pool and on individual connections.