Separate I2C mux state probing and gRPC forwarding state probing (#86)

What is the motivation for this PR?
Current implementation used the same APP DB table for active-active port forwarding state probing and active-standby port mux state probing. This is breaking CLI tests in pre-test.

How did you do it?
use tables below for active-active port forwarding state probing.
#define APP_FORWARDING_STATE_COMMAND_TABLE_NAME  "FORWARDING_STATE_COMMAND"
#define APP_FORWARDING_STATE_RESPONSE_TABLE_NAME "FORWARDING_STATE_RESPONSE"
RESPONSE table expects forwarding state response for both SELF and PEER.
Rename some table names & handler functions to avoid confusion.

How did you verify/test it?
Unit tests.

Author: zjswhhh

src/DbInterface.cpp

@@ -132,6 +132,23 @@ void DbInterface::probeMuxState(const std::string &portName)
     )));
 }
 
+//
+// ---> probeForwardingState(const std::string &portName)
+//
+// trigger tranceiver daemon to read Fowarding state using gRPC
+//
+void DbInterface::probeForwardingState(const std::string &portName)
+{
+    MUXLOGDEBUG(portName);
+
+    boost::asio::io_service &ioService = mStrand.context();
+    ioService.post(mStrand.wrap(boost::bind(
+        &DbInterface::handleProbeForwardingState,
+        this,
+        portName
+    )));
+}
+
 //
 // ---> setMuxLinkmgrState(const std::string &portName, link_manager::ActiveStandbyStateMachine::Label label);
 //
@@ -251,11 +268,14 @@ void DbInterface::initialize()
         mAppDbMuxTablePtr = std::make_shared<swss::ProducerStateTable> (
             mAppDbPtr.get(), APP_MUX_CABLE_TABLE_NAME
         );
+        mAppDbPeerMuxTablePtr = std::make_shared<swss::ProducerStateTable> (
+            mAppDbPtr.get(), APP_PEER_HW_FORWARDING_STATE_TABLE_NAME
+        );
         mAppDbMuxCommandTablePtr = std::make_shared<swss::Table> (
             mAppDbPtr.get(), APP_MUX_CABLE_COMMAND_TABLE_NAME
         );
-        mAppDbPeerMuxCommandTablePtr = std::make_shared<swss::Table> (
-            mAppDbPtr.get(), PEER_FORWARDING_STATE_COMMAND_TABLE
+        mAppDbForwardingCommandTablePtr = std::make_shared<swss::Table> (
+            mAppDbPtr.get(), APP_FORWARDING_STATE_COMMAND_TABLE_NAME
         );
         mStateDbMuxLinkmgrTablePtr = std::make_shared<swss::Table> (
             mStateDbPtr.get(), STATE_MUX_LINKMGR_TABLE_NAME
@@ -352,7 +372,7 @@ void DbInterface::handleSetPeerMuxState(const std::string portName, mux_state::M
         std::vector<swss::FieldValueTuple> values = {
             {"state", mMuxState[label]},
         };
-        mAppDbPeerMuxCommandTablePtr->set(portName, values);
+        mAppDbPeerMuxTablePtr->set(portName, values);
     }
 }
 
@@ -368,6 +388,18 @@ void DbInterface::handleProbeMuxState(const std::string portName)
     mAppDbMuxCommandTablePtr->hset(portName, "command", "probe");
 }
 
+//
+// ---> handleProbeForwardingState(const std::string portName)
+//
+// trigger xcvrd to read forwarding state using gRPC
+//
+void DbInterface::handleProbeForwardingState(const std::string portName)
+{
+    MUXLOGDEBUG(portName);
+
+    mAppDbForwardingCommandTablePtr->hset(portName, "command", "probe");
+}
+
 //
 // ---> handleSetMuxLinkmgrState(const std::string portName, link_manager::ActiveStandbyStateMachine::Label label);
 //
@@ -988,6 +1020,56 @@ void DbInterface::processMuxResponseNotifiction(std::deque<swss::KeyOpFieldsValu
     }
 }
 
+//
+// ---> processForwardingResponseNotification(std::deque<swss::KeyOpFieldsValuesTuple> &entries);
+//
+// process forwarding response (from xcvrd probing) notification
+//
+void DbInterface::processForwardingResponseNotification(std::deque<swss::KeyOpFieldsValuesTuple> &entries)
+{
+    for (auto &entry: entries) {
+        std::string port = kfvKey(entry);
+        std::string oprtation = kfvOp(entry);
+        std::vector<swss::FieldValueTuple> fieldValues = kfvFieldsValues(entry);
+
+        std::vector<swss::FieldValueTuple>::const_iterator cit_self = std::find_if(
+            fieldValues.cbegin(),
+            fieldValues.cend(),
+            [] (const swss::FieldValueTuple &fv) {return fvField(fv) == "response";}
+        );
+        if (cit_self != fieldValues.cend()) {
+            const std::string f = cit_self->first;
+            const std::string v = cit_self->second;
+
+            MUXLOGDEBUG(boost::format("port: %s, operation: %s, f: %s, v: %s") %
+                port %
+                oprtation %
+                f %
+                v
+            );
+            mMuxManagerPtr->processProbeMuxState(port, v);
+        }
+
+        std::vector<swss::FieldValueTuple>::const_iterator cit_peer = std::find_if(
+            fieldValues.cbegin(),
+            fieldValues.cend(),
+            [] (const swss::FieldValueTuple &fv) {return fvField(fv) == "response_peer";}
+        );
+        if (cit_peer != fieldValues.cend()) {
+            const std::string f = cit_peer->first;
+            const std::string v = cit_peer->second;
+
+            MUXLOGDEBUG(boost::format("port: %s, operation: %s, f: %s, v: %s") %
+                port %
+                oprtation %
+                f %
+                v
+            );
+            mMuxManagerPtr->processPeerMuxState(port, v);
+        }
+    }
+}
+
 //
 // ---> handleMuxResponseNotifiction(swss::SubscriberStateTable &appdbPortTable);
 //
@@ -1001,12 +1083,26 @@ void DbInterface::handleMuxResponseNotifiction(swss::SubscriberStateTable &appdb
     processMuxResponseNotifiction(entries);
 }
 
+//
+// ---> handleForwardingResponseNotification(swss::SubscriberStateTable &appdbForwardingResponseTable)
+//
+// handle fowarding response (from xcvrd) notification
+//
+void DbInterface::handleForwardingResponseNotification(swss::SubscriberStateTable &appdbForwardingResponseTable)
+{
+    std::deque<swss::KeyOpFieldsValuesTuple> entries;
+
+    appdbForwardingResponseTable.pops(entries);
+    processForwardingResponseNotification(entries);
+}
+
+
 //
 // ---> processPeerMuxResponseNotification(std::deque<swss::KeyOpFieldsValuesTuple> &entries);
 //
-// process peer MUX response (from xcvrd) notification
+// process peer MUX state (from xcvrd) notification
 //
-void DbInterface::processPeerMuxResponseNotification(std::deque<swss::KeyOpFieldsValuesTuple> &entries)
+void DbInterface::processPeerMuxNotification(std::deque<swss::KeyOpFieldsValuesTuple> &entries)
 {
     for (auto &entry: entries) {
         std::string port = kfvKey(entry);
@@ -1034,16 +1130,16 @@ void DbInterface::processPeerMuxResponseNotification(std::deque<swss::KeyOpField
 }
 
 //
-// ---> handlePeerMuxResponseNotification(swss::SubscriberStateTable &stateDbPeerMuxResponseTable);
+// ---> handlePeerMuxStateNotification(swss::SubscriberStateTable &stateDbPeerMuxTable);
 //
-// handles peer MUX response (from xcvrd) notification
+// handles peer MUX notification (from xcvrd) 
 //
-void DbInterface::handlePeerMuxResponseNotification(swss::SubscriberStateTable &stateDbPeerMuxResponseTable)
+void DbInterface::handlePeerMuxStateNotification(swss::SubscriberStateTable &stateDbPeerMuxTable)
 {
     std::deque<swss::KeyOpFieldsValuesTuple> entries;
 
-    stateDbPeerMuxResponseTable.pops(entries);
-    processPeerMuxResponseNotification(entries);
+    stateDbPeerMuxTable.pops(entries);
+    processPeerMuxNotification(entries);
 }
 
 //
@@ -1164,14 +1260,16 @@ void DbInterface::handleSwssNotification()
     swss::SubscriberStateTable appDbPortTable(appDbPtr.get(), APP_PORT_TABLE_NAME);
     // for command responses from the driver
     swss::SubscriberStateTable appDbMuxResponseTable(appDbPtr.get(), APP_MUX_CABLE_RESPONSE_TABLE_NAME);
+    // for command response of forwarding state probing from driver 
+    swss::SubscriberStateTable appDbForwardingResponseTable(appDbPtr.get(), APP_FORWARDING_STATE_RESPONSE_TABLE_NAME);
     // for getting state db MUX state when orchagent updates it
     swss::SubscriberStateTable stateDbPortTable(stateDbPtr.get(), STATE_MUX_CABLE_TABLE_NAME);
     // for getting state db default route state 
     swss::SubscriberStateTable stateDbRouteTable(stateDbPtr.get(), STATE_ROUTE_TABLE_NAME);
     // for getting peer's link status
     swss::SubscriberStateTable stateDbMuxInfoTable(stateDbPtr.get(), MUX_CABLE_INFO_TABLE);
     // for getting peer's admin forwarding state
-    swss::SubscriberStateTable stateDbPeerMuxResponseTable(stateDbPtr.get(), PEER_FORWARDING_STATE_RESPONSE_TABLE);
+    swss::SubscriberStateTable stateDbPeerMuxTable(stateDbPtr.get(), STATE_PEER_HW_FORWARDING_STATE_TABLE_NAME);
 
     getTorMacAddress(configDbPtr);
     getLoopback2InterfaceInfo(configDbPtr);
@@ -1192,10 +1290,11 @@ void DbInterface::handleSwssNotification()
     swssSelect.addSelectable(&configDbMuxTable);
     swssSelect.addSelectable(&appDbPortTable);
     swssSelect.addSelectable(&appDbMuxResponseTable);
+    swssSelect.addSelectable(&appDbForwardingResponseTable);
     swssSelect.addSelectable(&stateDbPortTable);
     swssSelect.addSelectable(&stateDbRouteTable);
     swssSelect.addSelectable(&stateDbMuxInfoTable);
-    swssSelect.addSelectable(&stateDbPeerMuxResponseTable);
+    swssSelect.addSelectable(&stateDbPeerMuxTable);
     swssSelect.addSelectable(&netlinkNeighbor);
 
     while (mPollSwssNotifcation) {
@@ -1220,14 +1319,16 @@ void DbInterface::handleSwssNotification()
             handleLinkStateNotifiction(appDbPortTable);
         } else if (selectable == static_cast<swss::Selectable *> (&appDbMuxResponseTable)) {
             handleMuxResponseNotifiction(appDbMuxResponseTable);
+        } else if (selectable == static_cast<swss::Selectable *> (&appDbForwardingResponseTable)) {
+            handleForwardingResponseNotification(appDbForwardingResponseTable);
         } else if (selectable == static_cast<swss::Selectable *> (&stateDbPortTable)) {
             handleMuxStateNotifiction(stateDbPortTable);
         } else if (selectable == static_cast<swss::Selectable *> (&stateDbRouteTable)) {
             handleDefaultRouteStateNotification(stateDbRouteTable);
         } else if (selectable == static_cast<swss::Selectable *> (&stateDbMuxInfoTable)) {
             handlePeerLinkStateNotification(stateDbMuxInfoTable);
-        } else if (selectable == static_cast<swss::Selectable *> (&stateDbPeerMuxResponseTable)) {
-            handlePeerMuxResponseNotification(stateDbPeerMuxResponseTable);
+        } else if (selectable == static_cast<swss::Selectable *> (&stateDbPeerMuxTable)) {
+            handlePeerMuxStateNotification(stateDbPeerMuxTable);
         } else if (selectable == static_cast<swss::Selectable *> (&netlinkNeighbor)) {
             continue;
         } else {

src/DbInterface.h

@@ -46,8 +46,12 @@ namespace mux
 {
 #define MUX_CABLE_INFO_TABLE  "MUX_CABLE_INFO"
 #define LINK_PROBE_STATS_TABLE_NAME "LINK_PROBE_STATS"
-#define PEER_FORWARDING_STATE_COMMAND_TABLE "HW_FORWARDING_STATE_PEER"
-#define PEER_FORWARDING_STATE_RESPONSE_TABLE "HW_MUX_CABLE_TABLE_PEER"
+
+#define APP_FORWARDING_STATE_COMMAND_TABLE_NAME  "FORWARDING_STATE_COMMAND"
+#define APP_FORWARDING_STATE_RESPONSE_TABLE_NAME "FORWARDING_STATE_RESPONSE"
+
+#define APP_PEER_HW_FORWARDING_STATE_TABLE_NAME    "HW_FORWARDING_STATE_PEER"
+#define STATE_PEER_HW_FORWARDING_STATE_TABLE_NAME   "HW_MUX_CABLE_TABLE_PEER"
 
 class MuxManager;
 using ServerIpPortMap = std::map<boost::asio::ip::address, std::string>;
@@ -158,6 +162,17 @@ class DbInterface
     */
     virtual void probeMuxState(const std::string &portName);
 
+    /**
+     * @method probeForwardingState
+     * 
+     * @brief  trigger tranceiver daemon to read Fowarding state using gRPC
+     * 
+     * @param portName (in) MUX/port name 
+     * 
+     * @return none
+     */
+    virtual void probeForwardingState(const std::string &portName);
+
     /**
     *@method setMuxLinkmgrState
     *
@@ -308,6 +323,17 @@ class DbInterface
     */
     void handleProbeMuxState(const std::string portName);
 
+    /**
+     * @method handleProbeForwardingState
+     * 
+     * @brief trigger xcvrd to read forwarding state using gRPC
+     * 
+     * @param portName (in) MUX/port name
+     * 
+     * @return none
+     */
+    void handleProbeForwardingState(const std::string portName);
+
     /**
     *@method handleSetMuxLinkmgrState
     *
@@ -582,6 +608,17 @@ class DbInterface
     */
     inline void processMuxResponseNotifiction(std::deque<swss::KeyOpFieldsValuesTuple> &entries);
 
+    /**
+     * @method processForwardingResponseNotification
+     * 
+     * @brief process forwarding response (from xcvrd probing) notification
+     * 
+     * @param entries (in) reference to app db entries
+     * 
+     * @return none
+     */
+    inline void processForwardingResponseNotification(std::deque<swss::KeyOpFieldsValuesTuple> &entries);
+
     /**
     *@method handleMuxResponseNotifiction
     *
@@ -594,26 +631,33 @@ class DbInterface
     void handleMuxResponseNotifiction(swss::SubscriberStateTable &appdbPortTable);
 
     /**
-    *@method processPeerMuxResponseNotification
+     * @method handleForwardingResponseNotification
+     * 
+     * @brief handle forwarding state response (from xcvrd probing) notification
+     */
+    void handleForwardingResponseNotification(swss::SubscriberStateTable &appdbForwardingResponseTable);
+
+    /**
+    *@method processPeerMuxNotification
     *
-    *@brief process peer MUX response (from xcvrd) notification
+    *@brief process peer MUX state (from xcvrd) notification
     *
-    *@param entries (in) reference to app db peer mux response table entries
+    *@param entries (in) reference to state db peer mux table entries
     *
     *@return none
     */
-    inline void processPeerMuxResponseNotification(std::deque<swss::KeyOpFieldsValuesTuple> &entries);
+    inline void processPeerMuxNotification(std::deque<swss::KeyOpFieldsValuesTuple> &entries);
 
     /**
-    *@method handlePeerMuxResponseNotification
+    *@method handlePeerMuxStateNotification
     *
-    *@brief handles peer MUX response (from xcvrd) notification
+    *@brief handles peer MUX notification (from xcvrd) 
     *
-    *@param appdbPortTable (in) reference to app db peer mux response table
+    *@param stateDbPeerMuxTable (in) reference to state db peer hw forwarding table 
     *
     *@return none
     */
-    void handlePeerMuxResponseNotification(swss::SubscriberStateTable &stateDbPeerMuxResponseTable);
+    void handlePeerMuxStateNotification(swss::SubscriberStateTable &stateDbPeerMuxTable);
 
     /**
     *@method processMuxStateNotifiction
@@ -684,10 +728,12 @@ class DbInterface
 
     // for communicating with orchagent
     std::shared_ptr<swss::ProducerStateTable> mAppDbMuxTablePtr;
+    // for communication with driver (setting peer's forwarding state)
+    std::shared_ptr<swss::ProducerStateTable> mAppDbPeerMuxTablePtr;
     // for communicating with the driver (probing the mux)
     std::shared_ptr<swss::Table> mAppDbMuxCommandTablePtr;
-    // for communicating with xcvrd to set peer mux state
-    std::shared_ptr<swss::Table> mAppDbPeerMuxCommandTablePtr;
+    // for communication with the driver (probing forwarding state)
+    std::shared_ptr<swss::Table> mAppDbForwardingCommandTablePtr;
     // for writing the current mux linkmgr health
     std::shared_ptr<swss::Table> mStateDbMuxLinkmgrTablePtr;
     // for writing mux metrics

src/MuxPort.cpp

@@ -364,4 +364,23 @@ void MuxPort::resetPckLossCount()
     )));
 }
 
+//
+// ---> probeMuxState()
+//
+// trigger xcvrd to read MUX state using i2c
+//
+void MuxPort::probeMuxState()
+{
+    switch (mMuxPortConfig.getPortCableType()) {
+        case common::MuxPortConfig::PortCableType::ActiveActive:
+            mDbInterfacePtr->probeForwardingState(mMuxPortConfig.getPortName());
+            break;
+        case common::MuxPortConfig::PortCableType::ActiveStandby:
+            mDbInterfacePtr->probeMuxState(mMuxPortConfig.getPortName());
+            break;
+        default:
+            break;
+    }
+}
+
 } /* namespace mux */

src/MuxPort.h

@@ -145,7 +145,7 @@ class MuxPort: public std::enable_shared_from_this<MuxPort>
     *
     *@return label of MUX state
     */
-    inline void probeMuxState() {mDbInterfacePtr->probeMuxState(mMuxPortConfig.getPortName());};
+    void probeMuxState();
 
     /**
     *@method setMuxLinkmgrState

test/FakeDbInterface.cpp

@@ -60,6 +60,11 @@ void FakeDbInterface::probeMuxState(const std::string &portName)
     mProbeMuxStateInvokeCount++;
 }
 
+void FakeDbInterface::probeForwardingState(const std::string &portName)
+{
+    mProbeForwardingStateInvokeCount++;
+}
+
 void FakeDbInterface::setMuxLinkmgrState(const std::string &portName, link_manager::ActiveStandbyStateMachine::Label label)
 {
     mSetMuxLinkmgrStateInvokeCount++;