pcied: Redis Performance Improvements

sonic-pcied/scripts/pcied

@@ -24,7 +24,6 @@ SIGNALS_TO_NAMES_DICT = dict((getattr(signal, n), n)
 
 SYSLOG_IDENTIFIER = "pcied"
 
-PCIE_RESULT_REGEX = "PCIe Device Checking All Test"
 PCIE_DEVICE_TABLE_NAME = "PCIE_DEVICE"
 PCIE_STATUS_TABLE_NAME = "PCIE_DEVICES"
 PCIE_DETACH_INFO_TABLE = "PCIE_DETACH_INFO"
@@ -88,12 +87,13 @@ class DaemonPcied(daemon_base.DaemonBase):
 
         self.timeout = PCIED_MAIN_THREAD_SLEEP_SECS
         self.stop_event = threading.Event()
-        self.state_db = None
+        self.redisPipeline = None
         self.device_table = None
         self.status_table = None
         self.resultInfo = []
         self.device_name = None
         self.aer_stats = {}
+        self.pcied_cache = {}
 
         global platform_pcieutil
 
@@ -103,10 +103,11 @@ class DaemonPcied(daemon_base.DaemonBase):
 
         # Connect to STATE_DB and create pcie device table
         try:
-            self.state_db = daemon_base.db_connect("STATE_DB")
-            self.device_table = swsscommon.Table(self.state_db, PCIE_DEVICE_TABLE_NAME)
-            self.status_table = swsscommon.Table(self.state_db, PCIE_STATUS_TABLE_NAME)
-            self.detach_info = swsscommon.Table(self.state_db, PCIE_DETACH_INFO_TABLE)
+            state_db = daemon_base.db_connect("STATE_DB")
+            self.redisPipeline = swsscommon.RedisPipeline(state_db,10)
+            self.device_table = swsscommon.Table(self.redisPipeline, PCIE_DEVICE_TABLE_NAME, True)
+            self.status_table = swsscommon.Table(self.redisPipeline, PCIE_STATUS_TABLE_NAME, True)
+            self.detach_info = swsscommon.Table(self.redisPipeline, PCIE_DETACH_INFO_TABLE, True)
         except Exception as e:
             log.log_error("Failed to connect to STATE_DB or create table. Error: {}".format(str(e)), True)
             sys.exit(PCIEUTIL_CONF_FILE_ERROR)
@@ -128,6 +129,7 @@ class DaemonPcied(daemon_base.DaemonBase):
     def update_aer_to_statedb(self):
         if self.aer_stats is None:
             self.log_debug("PCIe device {} has no AER Stats".format(self.device_name))
+            self.pcied_cache[self.device_name].pop("aer_stats", None)
             return
 
         try:
@@ -138,8 +140,10 @@ class DaemonPcied(daemon_base.DaemonBase):
             }
             if aer_fields:
                 self.device_table.set(self.device_name, swsscommon.FieldValuePairs(list(aer_fields.items())))
+                self.pcied_cache[self.device_name]["aer_stats"] = self.aer_stats
             else:
                 self.log_debug("PCIe device {} has no AER attributes".format(self.device_name))
+                self.pcied_cache[self.device_name].pop("aer_stats", None)
         except Exception as e:
             self.log_error("Exception while updating AER attributes to STATE_DB for {}: {}".format(self.device_name, str(e)))
 
@@ -148,11 +152,30 @@ class DaemonPcied(daemon_base.DaemonBase):
     def check_n_update_pcie_aer_stats(self, Bus, Dev, Fn):
         try:
             self.device_name = "%02x:%02x.%d" % (Bus, Dev, Fn)
-            Id = read_id_file(self.device_name)
+            current_id = read_id_file(self.device_name)
+            cached_device_data = self.pcied_cache.get(self.device_name, {})
+
+            if current_id is None:
+                if self.device_name in self.pcied_cache:
+                    del self.pcied_cache[self.device_name]
+                return
+
+            cached_id = cached_device_data.get("device_id", None)
+            id_changed = (cached_id != current_id)
+            # Check if id changed or adding id for the first time
+            if id_changed:
+                self.device_table.set(self.device_name, swsscommon.FieldValuePairs([('id', current_id)]))
+                # Ensure cache entry exists
+                if self.device_name not in self.pcied_cache:
+                    self.pcied_cache[self.device_name] = {}
+                self.pcied_cache[self.device_name]["device_id"] = current_id
+
             self.aer_stats = {}
-            if Id is not None:
-                self.device_table.set(self.device_name, swsscommon.FieldValuePairs([('id', Id)]))
-                self.aer_stats = platform_pcieutil.get_pcie_aer_stats(bus=Bus, dev=Dev, func=Fn)
+            self.aer_stats = platform_pcieutil.get_pcie_aer_stats(bus=Bus, dev=Dev, func=Fn)
+            cached_aer_stats = cached_device_data.get("aer_stats", None)
+            stats_changed = (cached_aer_stats != self.aer_stats)
+            # Only write to DB if AER stats changed
+            if stats_changed:
                 self.update_aer_to_statedb()
         except Exception as e:
             self.log_error("Exception while checking AER attributes for {}: {}".format(self.device_name, str(e)))
@@ -224,6 +247,11 @@ class DaemonPcied(daemon_base.DaemonBase):
 
         # update PCIe Device Status to DB
         self.update_pcie_devices_status_db(err)
+        # flush any remaining requests on the pipeline at end of every cycle
+        try:
+            self.redisPipeline.flush()
+        except Exception as e:
+            self.log_error(f"Exception occurred while flushing Redis pipeline : {type(e).__name__} : {e}")
 
    # Override signal handler from DaemonBase
     def signal_handler(self, sig, frame):

sonic-pcied/tests/mocked_libs/swsscommon/swsscommon.py

@@ -5,8 +5,21 @@
 STATE_DB = ''
 
 
+class RedisPipeline:
+    def __init__(self, db, batch_size=128):
+        self.db = db
+        self.batch_size = batch_size
+        self.queue = []
+
+    def flush(self):
+        # Mock flush operation - just clear the queue
+        self.queue.clear()
+        pass
+
+
 class Table:
-    def __init__(self, db, table_name):
+    def __init__(self, db_or_pipeline, table_name, buffered=False):
+        # Mock to support both both constructors (db, table_name) and (pipeline, table_name, buffered)
         self.table_name = table_name
         self.mock_dict = {}