[fpmsyncd] support flush with a timer

Author: a114j0y

fpmsyncd/fpmsyncd.cpp

@@ -9,13 +9,35 @@
 #include "subscriberstatetable.h"
 #include "warmRestartHelper.h"
 #include "fpmsyncd/fpmlink.h"
+#include "fpmsyncd/fpmsyncd.h"
 #include "fpmsyncd/routesync.h"
 
 #include <netlink/route/route.h>
 
 using namespace std;
 using namespace swss;
 
+// SELECT_TIMEOUT specifies the maximum wait time in milliseconds (-1 == infinite)
+static int SELECT_TIMEOUT;
+#define INFINITE -1
+#define FLUSH_TIMEOUT 500  // 500 milliseconds
+static int gFlushTimeout = FLUSH_TIMEOUT;
+// consider the traffic is small if pipeline contains < 500 entries
+#define SMALL_TRAFFIC 500
+
+/**
+ * @brief fpmsyncd invokes redispipeline's flush with a timer
+ * 
+ * redispipeline would automatically flush itself when full,
+ * but fpmsyncd can invoke pipeline's flush even if it's not full yet.
+ * 
+ * By setting SELECT_TIMEOUT, fpmsyncd controls the flush interval.
+ * 
+ * @param pipeline reference to the pipeline to be flushed
+ * @param scheduled if true, timer for fpmsyncd flush expired
+ */
+void flushPipeline(RedisPipeline& pipeline, bool scheduled);
+
 /*
  * Default warm-restart timer interval for routing-stack app. To be used only if
  * no explicit value has been defined in configuration.
@@ -61,7 +83,7 @@ int main(int argc, char **argv)
     DBConnector applStateDb("APPL_STATE_DB", 0);
     std::unique_ptr<NotificationConsumer> routeResponseChannel;
 
-    RedisPipeline pipeline(&db);
+    RedisPipeline pipeline(&db, ROUTE_SYNC_PPL_SIZE);
     RouteSync sync(&pipeline);
 
     DBConnector stateDb("STATE_DB", 0);
@@ -152,12 +174,14 @@ int main(int argc, char **argv)
                 sync.m_warmStartHelper.setState(WarmStart::WSDISABLED);
             }
 
+            SELECT_TIMEOUT = INFINITE;
+
             while (true)
             {
                 Selectable *temps;
 
                 /* Reading FPM messages forever (and calling "readMe" to read them) */
-                s.select(&temps);
+                auto ret = s.select(&temps, SELECT_TIMEOUT);
 
                 /*
                  * Upon expiration of the warm-restart timer or eoiu Hold Timer, proceed to run the
@@ -286,8 +310,7 @@ int main(int argc, char **argv)
                 }
                 else if (!warmStartEnabled || sync.m_warmStartHelper.isReconciled())
                 {
-                    pipeline.flush();
-                    SWSS_LOG_DEBUG("Pipeline flushed");
+                    flushPipeline(pipeline, ret==Select::TIMEOUT);
                 }
             }
         }
@@ -304,3 +327,30 @@ int main(int argc, char **argv)
 
     return 1;
 }
+
+void flushPipeline(RedisPipeline& pipeline, bool scheduled) {
+
+    size_t remaining = pipeline.size();
+
+    if (remaining == 0) {
+        SELECT_TIMEOUT = INFINITE;
+        return;
+    }
+
+    int idle = pipeline.getIdleTime();
+
+    // flush right away
+    if (remaining < SMALL_TRAFFIC || idle >= gFlushTimeout || idle <= 0 || scheduled) {
+
+        pipeline.flush();
+
+        SELECT_TIMEOUT = INFINITE;
+
+        SWSS_LOG_DEBUG("Pipeline flushed");
+    
+        return;
+    }
+
+    // postpone the flush
+    SELECT_TIMEOUT = gFlushTimeout - idle;
+}

fpmsyncd/fpmsyncd.h

@@ -0,0 +1,7 @@
+#ifndef __FPMSYNCD__
+#define __FPMSYNCD__
+
+// redispipeline has a maximum capacity of 50000 entries
+#define ROUTE_SYNC_PPL_SIZE 50000
+
+#endif

fpmsyncd/routesync.cpp

@@ -76,7 +76,7 @@ static decltype(auto) makeNlAddr(const T& ip)
 
 
 RouteSync::RouteSync(RedisPipeline *pipeline) :
-    m_routeTable(pipeline, APP_ROUTE_TABLE_NAME, true),
+    m_routeTable(pipeline, APP_ROUTE_TABLE_NAME, true, true),
     m_label_routeTable(pipeline, APP_LABEL_ROUTE_TABLE_NAME, true),
     m_vnet_routeTable(pipeline, APP_VNET_RT_TABLE_NAME, true),
     m_vnet_tunnelTable(pipeline, APP_VNET_RT_TUNNEL_TABLE_NAME, true),