Warm reboot for PortsOrch

Signed-off-by: Qi Luo <[email protected]>

Author: qiluo-msft

orchagent/orch.cpp

@@ -66,13 +66,10 @@ vector<Selectable *> Orch::getSelectables()
     return selectables;
 }
 
-void Consumer::execute()
+void Consumer::addToSync(std::deque<KeyOpFieldsValuesTuple> &entries)
 {
     SWSS_LOG_ENTER();
 
-    std::deque<KeyOpFieldsValuesTuple> entries;
-    getConsumerTable()->pops(entries);
-
     /* Nothing popped */
     if (entries.empty())
     {
@@ -123,6 +120,16 @@ void Consumer::execute()
             m_toSync[key] = KeyOpFieldsValuesTuple(key, op, existing_values);
         }
     }
+}
+
+void Consumer::execute()
+{
+    SWSS_LOG_ENTER();
+
+    std::deque<KeyOpFieldsValuesTuple> entries;
+    getConsumerTable()->pops(entries);
+
+    addToSync(entries);
 
     drain();
 }
@@ -133,6 +140,48 @@ void Consumer::drain()
         m_orch->doTask(*this);
 }
 
+// TODO: Table should be const
+void Orch::addExistingData(Table *table)
+{
+    string tableName = table->getTableName();
+    Consumer* consumer;
+    auto it = m_consumerMap.begin();
+
+    while (it != m_consumerMap.end())
+    {
+        consumer = (Consumer*)(it->second.get());
+        if (tableName == consumer->getTableName())
+        {
+            break;
+        }
+        it++;
+    }
+
+    if (it == m_consumerMap.end())
+    {
+        return;
+    }
+
+    std::deque<KeyOpFieldsValuesTuple> entries;
+    vector<string> keys;
+    table->getKeys(keys);
+    for (const auto &key: keys)
+    {
+        KeyOpFieldsValuesTuple kco;
+
+        kfvKey(kco) = key;
+        kfvOp(kco) = SET_COMMAND;
+
+        if (!table->get(key, kfvFieldsValues(kco)))
+        {
+            continue;
+        }
+        entries.push_back(kco);
+    }
+
+    consumer->addToSync(entries);
+}
+
 /*
 - Validates reference has proper format which is [table_name:object_name]
 - validates table_name exists
@@ -364,6 +413,10 @@ void Orch::addConsumer(DBConnector *db, string tableName, int pri)
     {
         addExecutor(tableName, new Consumer(new SubscriberStateTable(db, tableName, TableConsumable::DEFAULT_POP_BATCH_SIZE, pri), this));
     }
+    else if (tableName == APP_PORT_TABLE_NAME)
+    {
+        addExecutor(tableName, new Consumer(new ConsumerTable(db, tableName, gBatchSize, pri), this));
+    }
     else
     {
         addExecutor(tableName, new Consumer(new ConsumerStateTable(db, tableName, gBatchSize, pri), this));

orchagent/orch.h

@@ -116,6 +116,7 @@ class Consumer : public Executor {
         return getConsumerTable()->getTableName();
     }
 
+    void addToSync(std::deque<KeyOpFieldsValuesTuple> &entries);
     void execute();
     void drain();
 
@@ -149,6 +150,9 @@ class Orch
 
     vector<Selectable*> getSelectables();
 
+    // add the existing data to the consumer todo task list.
+    void addExistingData(Table *table);
+
     /* Iterate all consumers in m_consumerMap and run doTask(Consumer) */
     void doTask();
 

orchagent/portsorch.cpp

@@ -245,6 +245,9 @@ PortsOrch::PortsOrch(DBConnector *db, vector<table_name_with_pri_t> &tableNames)
     m_portStatusNotificationConsumer = new swss::NotificationConsumer(notificationsDb, "NOTIFICATIONS");
     auto portStatusNotificatier = new Notifier(m_portStatusNotificationConsumer, this);
     Orch::addExecutor("PORT_STATUS_NOTIFICATIONS", portStatusNotificatier);
+
+    // Try warm start
+    bake();
 }
 
 void PortsOrch::removeDefaultVlanMembers()
@@ -1177,6 +1180,64 @@ bool PortsOrch::initPort(const string &alias, const set<int> &lane_set)
     return true;
 }
 
+bool PortsOrch::bake()
+{
+    SWSS_LOG_ENTER();
+
+    // Check the APP_DB port table for warm reboot
+    vector<FieldValueTuple> tuples;
+    bool foundPortConfigDone = m_portTable->get("PortConfigDone", tuples);
+    SWSS_LOG_NOTICE("foundPortConfigDone = %d", foundPortConfigDone);
+
+    bool foundPortInitDone = m_portTable->get("PortInitDone", tuples);
+    SWSS_LOG_NOTICE("foundPortInitDone = %d", foundPortInitDone);
+
+    vector<string> keys;
+    m_portTable->getKeys(keys);
+    SWSS_LOG_NOTICE("m_portTable->getKeys %zd", keys.size());
+
+    if (!foundPortConfigDone || !foundPortInitDone)
+    {
+        SWSS_LOG_NOTICE("No port table, fallback to cold start");
+        cleanPortTable(keys);
+        return false;
+    }
+
+    doPortConfigDoneTask(tuples);
+    if (m_portCount != keys.size() - 2)
+    {
+        // Invalid port table
+        SWSS_LOG_ERROR("Invalid port table: m_portCount");
+        cleanPortTable(keys);
+        return false;
+    }
+
+    addExistingData(m_portTable.get());
+    return true;
+}
+
+// Clean up port table
+void PortsOrch::cleanPortTable(const vector<string>& keys)
+{
+    for (auto& key : keys)
+    {
+        m_portTable->del(key);
+    }
+}
+
+void PortsOrch::doPortConfigDoneTask(const vector<FieldValueTuple>& tuples)
+{
+    m_portConfigDone = true;
+
+    for (auto i : tuples)
+    {
+        if (fvField(i) == "count")
+        {
+            m_portCount = to_uint<uint32_t>(fvValue(i));
+        }
+    }
+}
+
 void PortsOrch::doPortTask(Consumer &consumer)
 {
     SWSS_LOG_ENTER();
@@ -1191,15 +1252,7 @@ void PortsOrch::doPortTask(Consumer &consumer)
 
         if (alias == "PortConfigDone")
         {
-            m_portConfigDone = true;
-
-            for (auto i : kfvFieldsValues(t))
-            {
-                if (fvField(i) == "count")
-                {
-                    m_portCount = to_uint<uint32_t>(fvValue(i));
-                }
-            }
+            doPortConfigDoneTask(kfvFieldsValues(t));
         }
 
         /* Get notification from application */

orchagent/portsorch.h

@@ -56,6 +56,9 @@ class PortsOrch : public Orch, public Subject
     bool isInitDone();
 
     map<string, Port>& getAllPorts();
+    bool bake();
+    void cleanPortTable(const vector<string>& keys);
+    void doPortConfigDoneTask(const vector<FieldValueTuple>& tuples);
     bool getBridgePort(sai_object_id_t id, Port &port);
     bool getPort(string alias, Port &port);
     bool getPort(sai_object_id_t id, Port &port);

portsyncd/linksync.h

@@ -2,7 +2,7 @@
 #define __LINKSYNC__
 
 #include "dbconnector.h"
-#include "producerstatetable.h"
+#include "producertable.h"
 #include "netmsg.h"
 
 #include <map>
@@ -19,7 +19,7 @@ class LinkSync : public NetMsg
     virtual void onMsg(int nlmsg_type, struct nl_object *obj);
 
 private:
-    ProducerStateTable m_portTableProducer;
+    ProducerTable m_portTableProducer;
     Table m_portTable, m_statePortTable;
 
     std::map<unsigned int, std::string> m_ifindexNameMap;

portsyncd/portsyncd.cpp

@@ -11,6 +11,7 @@
 #include "netdispatcher.h"
 #include "netlink.h"
 #include "producerstatetable.h"
+#include "producertable.h"
 #include "portsyncd/linksync.h"
 #include "subscriberstatetable.h"
 #include "exec.h"
@@ -38,10 +39,11 @@ void usage()
     cout << "                           use configDB data if not specified" << endl;
 }
 
-void handlePortConfigFile(ProducerStateTable &p, string file);
-void handlePortConfigFromConfigDB(ProducerStateTable &p, DBConnector &cfgDb);
+// TODO: find a common base class for ProducerTable and ProducerStateTable
+void handlePortConfigFile(ProducerTable &p, string file);
+void handlePortConfigFromConfigDB(ProducerTable &p, DBConnector &cfgDb);
 void handleVlanIntfFile(string file);
-void handlePortConfig(ProducerStateTable &p, map<string, KeyOpFieldsValuesTuple> &port_cfg_map);
+void handlePortConfig(ProducerTable &p, map<string, KeyOpFieldsValuesTuple> &port_cfg_map);
 
 int main(int argc, char **argv)
 {
@@ -69,7 +71,7 @@ int main(int argc, char **argv)
     DBConnector cfgDb(CONFIG_DB, DBConnector::DEFAULT_UNIXSOCKET, 0);
     DBConnector appl_db(APPL_DB, DBConnector::DEFAULT_UNIXSOCKET, 0);
     DBConnector state_db(STATE_DB, DBConnector::DEFAULT_UNIXSOCKET, 0);
-    ProducerStateTable p(&appl_db, APP_PORT_TABLE_NAME);
+    ProducerTable p(&appl_db, APP_PORT_TABLE_NAME);
     SubscriberStateTable portCfg(&cfgDb, CFG_PORT_TABLE_NAME);
 
     LinkSync sync(&appl_db, &state_db);
@@ -157,15 +159,15 @@ int main(int argc, char **argv)
     return 1;
 }
 
-static void notifyPortConfigDone(ProducerStateTable &p)
+static void notifyPortConfigDone(ProducerTable &p)
 {
     /* Notify that all ports added */
     FieldValueTuple finish_notice("count", to_string(g_portSet.size()));
     vector<FieldValueTuple> attrs = { finish_notice };
     p.set("PortConfigDone", attrs);
 }
 
-void handlePortConfigFromConfigDB(ProducerStateTable &p, DBConnector &cfgDb)
+void handlePortConfigFromConfigDB(ProducerTable &p, DBConnector &cfgDb)
 {
     cout << "Get port configuration from ConfigDB..." << endl;
 
@@ -188,7 +190,7 @@ void handlePortConfigFromConfigDB(ProducerStateTable &p, DBConnector &cfgDb)
     notifyPortConfigDone(p);
 }
 
-void handlePortConfigFile(ProducerStateTable &p, string file)
+void handlePortConfigFile(ProducerTable &p, string file)
 {
     cout << "Read port configuration file..." << endl;
 
@@ -273,7 +275,7 @@ void handlePortConfigFile(ProducerStateTable &p, string file)
     notifyPortConfigDone(p);
 }
 
-void handlePortConfig(ProducerStateTable &p, map<string, KeyOpFieldsValuesTuple> &port_cfg_map)
+void handlePortConfig(ProducerTable &p, map<string, KeyOpFieldsValuesTuple> &port_cfg_map)
 {
 
     auto it = port_cfg_map.begin();