Warm reboot for general Orch (sonic-net#572)

* Warm reboot for BufferOrch
* Add log for warm reboot
* Add bake() interface
* OrchDaemon supports warm start

Author: qiluo-msft

orchagent/orch.cpp

@@ -66,14 +66,14 @@ vector<Selectable *> Orch::getSelectables()
     return selectables;
 }
 
-void Consumer::addToSync(std::deque<KeyOpFieldsValuesTuple> &entries)
+size_t Consumer::addToSync(std::deque<KeyOpFieldsValuesTuple> &entries)
 {
     SWSS_LOG_ENTER();
 
     /* Nothing popped */
     if (entries.empty())
     {
-        return;
+        return 0;
     }
 
     for (auto& entry: entries)
@@ -120,10 +120,11 @@ void Consumer::addToSync(std::deque<KeyOpFieldsValuesTuple> &entries)
             m_toSync[key] = KeyOpFieldsValuesTuple(key, op, existing_values);
         }
     }
+    return entries.size();
 }
 
 // TODO: Table should be const
-void Consumer::refillToSync(Table* table)
+size_t Consumer::refillToSync(Table* table)
 {
     std::deque<KeyOpFieldsValuesTuple> entries;
     vector<string> keys;
@@ -142,15 +143,28 @@ void Consumer::refillToSync(Table* table)
         entries.push_back(kco);
     }
 
-    addToSync(entries);
+    return addToSync(entries);
 }
 
-void Consumer::refillToSync()
+size_t Consumer::refillToSync()
 {
-    auto db = getConsumerTable()->getDbConnector();
-    string tableName = getConsumerTable()->getTableName();
-    auto table = Table(db, tableName);
-    refillToSync(&table);
+    ConsumerTableBase *consumerTable = getConsumerTable();
+
+    auto subTable = dynamic_cast<SubscriberStateTable *>(consumerTable);
+    if (subTable != NULL)
+    {
+        std::deque<KeyOpFieldsValuesTuple> entries;
+        subTable->pops(entries);
+        return addToSync(entries);
+    }
+    else
+    {
+        // consumerTable is either ConsumerStateTable or ConsumerTable
+        auto db = consumerTable->getDbConnector();
+        string tableName = consumerTable->getTableName();
+        auto table = Table(db, tableName);
+        return refillToSync(&table);
+    }
 }
 
 void Consumer::execute()
@@ -171,31 +185,50 @@ void Consumer::drain()
         m_orch->doTask(*this);
 }
 
-bool Orch::addExistingData(const string& tableName)
+size_t Orch::addExistingData(const string& tableName)
 {
-    Consumer* consumer = dynamic_cast<Consumer *>(getExecutor(tableName));
+    auto consumer = dynamic_cast<Consumer *>(getExecutor(tableName));
     if (consumer == NULL)
     {
         SWSS_LOG_ERROR("No consumer %s in Orch", tableName.c_str());
-        return false;
+        return 0;
     }
 
-    consumer->refillToSync();
-    return true;
+    return consumer->refillToSync();
 }
 
 // TODO: Table should be const
-bool Orch::addExistingData(Table *table)
+size_t Orch::addExistingData(Table *table)
 {
     string tableName = table->getTableName();
     Consumer* consumer = dynamic_cast<Consumer *>(getExecutor(tableName));
     if (consumer == NULL)
     {
         SWSS_LOG_ERROR("No consumer %s in Orch", tableName.c_str());
-        return false;
+        return 0;
+    }
+
+    return consumer->refillToSync(table);
+}
+
+bool Orch::bake()
+{
+    SWSS_LOG_ENTER();
+
+    for(auto &it : m_consumerMap)
+    {
+        string executorName = it.first;
+        auto executor = it.second;
+        auto consumer = dynamic_cast<Consumer *>(executor.get());
+        if (consumer == NULL)
+        {
+            continue;
+        }
+        
+        size_t refilled = consumer->refillToSync();
+        SWSS_LOG_NOTICE("Add warm input: %s, %zd", executorName.c_str(), refilled);
     }
 
-    consumer->refillToSync(table);
     return true;
 }
 

orchagent/orch.h

@@ -116,15 +116,18 @@ class Consumer : public Executor {
         return getConsumerTable()->getTableName();
     }
 
-    void addToSync(std::deque<KeyOpFieldsValuesTuple> &entries);
-    void refillToSync();
-    void refillToSync(Table* table);
+    size_t refillToSync();
+    size_t refillToSync(Table* table);
     void execute();
     void drain();
 
     /* Store the latest 'golden' status */
     // TODO: hide?
     SyncMap m_toSync;
+    
+protected:
+    // Returns: the number of entries added to m_toSync
+    size_t addToSync(std::deque<KeyOpFieldsValuesTuple> &entries);
 };
 
 typedef map<string, std::shared_ptr<Executor>> ConsumerMap;
@@ -153,9 +156,13 @@ class Orch
     vector<Selectable*> getSelectables();
 
     // add the existing table data (left by warm reboot) to the consumer todo task list.
-    bool addExistingData(Table *table);
-    bool addExistingData(const string& tableName);
+    size_t addExistingData(Table *table);
+    size_t addExistingData(const string& tableName);
 
+    // Prepare for warm start if Redis contains valid input data
+    // otherwise fallback to cold start
+    virtual bool bake();
+    
     /* Iterate all consumers in m_consumerMap and run doTask(Consumer) */
     void doTask();
 

orchagent/orchdaemon.cpp

@@ -274,6 +274,12 @@ void OrchDaemon::start()
 {
     SWSS_LOG_ENTER();
 
+    // Try warm start
+    for (Orch *o : m_orchList)
+    {
+        o->bake();
+    }
+
     for (Orch *o : m_orchList)
     {
         m_select->addSelectables(o->getSelectables());

orchagent/portsorch.cpp

@@ -245,9 +245,6 @@ 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()

orchagent/portsorch.h

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