psud: Adding Redis Pipeline for performance improvement

sonic-psud/scripts/psud

@@ -403,10 +403,11 @@ class DaemonPsud(daemon_base.DaemonBase):
 
         # Connect to STATE_DB and create psu/chassis info tables
         state_db = daemon_base.db_connect("STATE_DB")
-        self.chassis_tbl = swsscommon.Table(state_db, CHASSIS_INFO_TABLE)
-        self.psu_tbl = swsscommon.Table(state_db, PSU_INFO_TABLE)
-        self.fan_tbl = swsscommon.Table(state_db, FAN_INFO_TABLE)
-        self.phy_entity_tbl = swsscommon.Table(state_db, PHYSICAL_ENTITY_INFO_TABLE)
+        self.statedb_redisPipeline = swsscommon.RedisPipeline(state_db, 10)
+        self.chassis_tbl = swsscommon.Table(self.statedb_redisPipeline, CHASSIS_INFO_TABLE, True)
+        self.psu_tbl = swsscommon.Table(self.statedb_redisPipeline, PSU_INFO_TABLE, True)
+        self.fan_tbl = swsscommon.Table(self.statedb_redisPipeline, FAN_INFO_TABLE, True)
+        self.phy_entity_tbl = swsscommon.Table(self.statedb_redisPipeline, PHYSICAL_ENTITY_INFO_TABLE, True)
 
         # Post psu number info to STATE_DB
         self.num_psus = _wrapper_get_num_psus()
@@ -459,6 +460,11 @@ class DaemonPsud(daemon_base.DaemonBase):
         if self.first_run:
             self.first_run = False
 
+        # flush any remaining requests on the pipeline to STATE_DB at end of every cycle
+        try:
+            self.statedb_redisPipeline.flush()
+        except Exception as e:
+            self.log_error(f"Exception occurred while flushing Redis pipeline : {type(e).__name__} : {e}")
         return True
 
     def update_psu_data(self):

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

@@ -6,9 +6,19 @@
 
 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()
 
 class Table:
-    def __init__(self, db, table_name):
+    def __init__(self, db_or_pipeline, table_name, buffered=False):
+        # Mock to support both constructors (db, table_name) and (pipeline, table_name, buffered)
         self.table_name = table_name
         self.mock_dict = {}
 

sonic-psud/tests/test_DaemonPsud.py

@@ -103,9 +103,21 @@ def test_run(self):
         daemon_psud.update_psu_data = mock.MagicMock()
         daemon_psud._update_led_color = mock.MagicMock()
         daemon_psud.update_psu_chassis_info = mock.MagicMock()
+        daemon_psud.statedb_redisPipeline = mock.MagicMock()
 
         daemon_psud.run()
         assert daemon_psud.first_run is False
+        daemon_psud.statedb_redisPipeline.flush.assert_called_once()
+
+        # Test Redis pipeline flush exception scenario
+        daemon_psud.statedb_redisPipeline.flush.side_effect = Exception("Flush error")
+        daemon_psud.log_error = mock.MagicMock()
+
+        result = daemon_psud.run()
+        assert result is True
+        assert daemon_psud.first_run is False
+        daemon_psud.log_error.assert_called_once()
+        assert "Exception occurred while flushing Redis pipeline" in daemon_psud.log_error.call_args[0][0]
 
     def test_update_psu_data(self):
         mock_psu1 = MockPsu("PSU 1", 0, True, True)