Swss.orchagent.multi.consumer

orchagent/Makefile.am

@@ -10,7 +10,7 @@ else
 DBGFLAGS = -g
 endif
 
-orchagent_SOURCES = main.cpp orchdaemon.cpp orch.cpp routeorch.cpp neighorch.cpp intfsorch.cpp portsorch.cpp
+orchagent_SOURCES = main.cpp orchdaemon.cpp orch.cpp routeorch.cpp neighorch.cpp intfsorch.cpp portsorch.cpp qosorch.cpp
 
 orchagent_CFLAGS = $(DBGFLAGS) $(AM_CFLAGS) $(CFLAGS_COMMON) $(CFLAGS_SAI)
 orchagent_CPPFLAGS = $(DBGFLAGS) $(AM_CFLAGS) $(CFLAGS_COMMON) $(CFLAGS_SAI)

orchagent/intfsorch.cpp

@@ -18,13 +18,13 @@ IntfsOrch::IntfsOrch(DBConnector *db, string tableName, PortsOrch *portsOrch) :
 {
 }
 
-void IntfsOrch::doTask()
+void IntfsOrch::doTask(_in_ Consumer& consumer_info)
 {
-    if (m_toSync.empty())
+    if (consumer_info.m_toSync.empty())
         return;
 
-    auto it = m_toSync.begin();
-    while (it != m_toSync.end())
+    auto it = consumer_info.m_toSync.begin();
+    while (it != consumer_info.m_toSync.end())
     {
         KeyOpFieldsValuesTuple t = it->second;
 
@@ -33,15 +33,15 @@ void IntfsOrch::doTask()
         if (found == string::npos)
         {
             SWSS_LOG_ERROR("Failed to parse task key %s\n", key.c_str());
-            it = m_toSync.erase(it);
+            it = consumer_info.m_toSync.erase(it);
             continue;
         }
         string alias = key.substr(0, found);
 
         IpPrefix ip_prefix(key.substr(found+1));
         if (!ip_prefix.isV4())
         {
-            it = m_toSync.erase(it);
+            it = consumer_info.m_toSync.erase(it);
             continue;
         }
 
@@ -52,15 +52,15 @@ void IntfsOrch::doTask()
             /* Duplicate entry */
             if (m_intfs.find(alias) != m_intfs.end() && m_intfs[alias] == ip_prefix.getIp())
             {
-                it = m_toSync.erase(it);
+                it = consumer_info.m_toSync.erase(it);
                 continue;
             }
 
             Port p;
             if (!m_portsOrch->getPort(alias, p))
             {
                 SWSS_LOG_ERROR("Failed to locate interface %s\n", alias.c_str());
-                it = m_toSync.erase(it);
+                it = consumer_info.m_toSync.erase(it);
                 continue;
             }
 
@@ -110,7 +110,7 @@ void IntfsOrch::doTask()
             {
                 SWSS_LOG_NOTICE("Create packet action trap route ip:%s\n", ip_prefix.getIp().to_string().c_str());
                 m_intfs[alias] = ip_prefix.getIp();
-                it = m_toSync.erase(it);
+                it = consumer_info.m_toSync.erase(it);
             }
         }
         else if (op == DEL_COMMAND)
@@ -119,7 +119,7 @@ void IntfsOrch::doTask()
             if (!m_portsOrch->getPort(alias, p))
             {
                 SWSS_LOG_ERROR("Failed to locate interface %s\n", alias.c_str());
-                it = m_toSync.erase(it);
+                it = consumer_info.m_toSync.erase(it);
                 continue;
             }
 
@@ -152,7 +152,7 @@ void IntfsOrch::doTask()
             {
                 SWSS_LOG_NOTICE("Remove packet action trap route ip:%s\n", ip_prefix.getIp().to_string().c_str());
                 m_intfs.erase(alias);
-                it = m_toSync.erase(it);
+                it = consumer_info.m_toSync.erase(it);
             }
         }
     }

orchagent/intfsorch.h

@@ -16,8 +16,7 @@ class IntfsOrch : public Orch
 public:
     IntfsOrch(DBConnector *db, string tableName, PortsOrch *portsOrch);
 private:
-    void doTask();
-
+    virtual void doTask(_in_ Consumer& consumer_info);
     PortsOrch *m_portsOrch;
     IntfsTable m_intfs;
 };

orchagent/neighorch.cpp

@@ -5,37 +5,37 @@
 extern sai_neighbor_api_t*         sai_neighbor_api;
 extern sai_next_hop_api_t*         sai_next_hop_api;
 
-void NeighOrch::doTask()
+void NeighOrch::doTask(_in_ Consumer& consumer_info)
 {
-    if (m_toSync.empty())
+    if (consumer_info.m_toSync.empty())
         return;
 
-    auto it = m_toSync.begin();
-    while (it != m_toSync.end())
+    auto it = consumer_info.m_toSync.begin();
+    while (it != consumer_info.m_toSync.end())
     {
         KeyOpFieldsValuesTuple t = it->second;
 
         string key = kfvKey(t);
-        size_t found = key.find(':');
+        size_t found = key.find(delimiter);
         if (found == string::npos)
         {
             SWSS_LOG_ERROR("Failed to parse task key %s\n", key.c_str());
-            it = m_toSync.erase(it);
+            it = consumer_info.m_toSync.erase(it);
             continue;
         }
         string alias = key.substr(0, found);
         Port p;
 
         if (!m_portsOrch->getPort(alias, p))
         {
-            it = m_toSync.erase(it);
+            it = consumer_info.m_toSync.erase(it);
             continue;
         }
 
         IpAddress ip_address(key.substr(found+1));
         if (!ip_address.isV4())
         {
-            it = m_toSync.erase(it);
+            it = consumer_info.m_toSync.erase(it);
             continue;
         }
 
@@ -49,37 +49,37 @@ void NeighOrch::doTask()
             for (auto i = kfvFieldsValues(t).begin();
                  i  != kfvFieldsValues(t).end(); i++)
             {
-                if (fvField(*i) == "neigh")
+                if (fvField(*i) == neigh_field_name)
                     mac_address = MacAddress(fvValue(*i));
             }
 
             if (m_syncdNeighbors.find(neighbor_entry) == m_syncdNeighbors.end() || m_syncdNeighbors[neighbor_entry] != mac_address)
             {
                 if (addNeighbor(neighbor_entry, mac_address))
-                    it = m_toSync.erase(it);
+                    it = consumer_info.m_toSync.erase(it);
                 else
                     it++;
             }
             else
-                it = m_toSync.erase(it);
+                it = consumer_info.m_toSync.erase(it);
         }
         else if (op == DEL_COMMAND)
         {
             if (m_syncdNeighbors.find(neighbor_entry) != m_syncdNeighbors.end())
             {
                 if (removeNeighbor(neighbor_entry))
-                    it = m_toSync.erase(it);
+                    it = consumer_info.m_toSync.erase(it);
                 else
                     it++;
             }
             /* Cannot locate the neighbor */
             else
-                it = m_toSync.erase(it);
+                it = consumer_info.m_toSync.erase(it);
         }
         else
         {
             SWSS_LOG_ERROR("Unknown operation type %s\n", op.c_str());
-            it = m_toSync.erase(it);
+            it = consumer_info.m_toSync.erase(it);
         }
     }
 }
@@ -143,8 +143,8 @@ bool NeighOrch::addNeighbor(NeighborEntry neighborEntry, MacAddress macAddress)
     }
     else
     {
-        // XXX: The neighbor entry is already there
-        // XXX: MAC change
+        // TODO: The neighbor entry is already there
+        // TODO: MAC change
     }
 
     return true;

orchagent/neighorch.h

@@ -23,14 +23,15 @@ typedef map<NeighborEntry, MacAddress> NeighborTable;
 class NeighOrch : public Orch
 {
 public:
-
     NeighOrch(DBConnector *db, string tableName, PortsOrch *portsOrch, RouteOrch *routeOrch) :
         Orch(db, tableName), m_portsOrch(portsOrch), m_routeOrch(routeOrch) {};
+    const char* const delimiter         = ";";
+    const char* const neigh_field_name  = "neigh";
 private:
     PortsOrch *m_portsOrch;
     RouteOrch *m_routeOrch;
 
-    void doTask();
+    virtual void doTask(_in_ Consumer& consumer_info);
 
     NeighborTable m_syncdNeighbors;
 

orchagent/orch.cpp

@@ -1,66 +1,105 @@
 #include "orch.h"
-
 #include "logger.h"
 
+using namespace swss;
+
 Orch::Orch(DBConnector *db, string tableName) :
-    m_db(db), m_name(tableName)
+    m_db(db)
 {
-    m_consumer = new ConsumerTable(m_db, tableName);
+    Consumer c_info(new ConsumerTable(m_db, tableName));
+    m_consumer_map.insert(ConsumerMapPair(tableName, c_info));
+}
+
+Orch::Orch(DBConnector *db, vector<string> &tableNames) :
+    m_db(db)
+{
+    for( auto it = tableNames.begin(); it != tableNames.end(); it++) {
+        Consumer c_info(new ConsumerTable(m_db, *it));
+        m_consumer_map.insert(ConsumerMapPair(*it, c_info));
+    }
 }
 
 Orch::~Orch()
 {
     delete(m_db);
-    delete(m_consumer);
+    for(auto it : m_consumer_map) {
+        delete it.second.m_consumer;
+    }
 }
 
-void Orch::execute()
+void Orch::getSelectables(_out_ std::vector<Selectable*> &consumers)
 {
-    KeyOpFieldsValuesTuple t;
-    m_consumer->pop(t);
+    consumers.clear();
+    for(auto it : m_consumer_map) {
+        consumers.push_back(it.second.m_consumer);
+    }
+}
+
+bool Orch::is_owned_consumer(ConsumerTable*consumer) const
+{
+    for(auto it : m_consumer_map) {
+        if(it.second.m_consumer == consumer) {
+            return true;
+        }
+    }
+    return false;
+}
+
+bool Orch::execute(string tableName)
+{
+    auto consumer_it = m_consumer_map.find(tableName);
+    if(consumer_it == m_consumer_map.end()) {
+        SWSS_LOG_ERROR("Unrecognized tableName:%s\n", tableName.c_str());
+        return false;
+    }
+    Consumer& consumer = consumer_it->second;
+
+    KeyOpFieldsValuesTuple new_data;
+    consumer.m_consumer->pop(new_data);
 
-    string key = kfvKey(t);
-    string op  = kfvOp(t);
+    string key = kfvKey(new_data);
+    string op  = kfvOp(new_data);
 
 #ifdef DEBUG
-    string debug = "Orch : " + m_name + " key : " + kfvKey(t) + " op : "  + kfvOp(t);
-    for (auto i = kfvFieldsValues(t).begin(); i != kfvFieldsValues(t).end(); i++)
+    string debug = "Table : " + consumer.m_consumer.getTableName() + " key : " + kfvKey(new_data) + " op : "  + kfvOp(new_data);
+    for (auto i = kfvFieldsValues(new_data).begin(); i != kfvFieldsValues(new_data).end(); i++)
         debug += " " + fvField(*i) + " : " + fvValue(*i);
     SWSS_LOG_DEBUG("%s\n", debug.c_str());
 #endif
 
-    /* If a new task comes or if a DEL task comes, we directly put it into m_toSync map */
-    if ( m_toSync.find(key) == m_toSync.end() || op == DEL_COMMAND)
+    /* If a new task comes or if a DEL task comes, we directly put it into consumer.m_toSync map */
+    if ( consumer.m_toSync.find(key) == consumer.m_toSync.end() || op == DEL_COMMAND)
     {
-       m_toSync[key] = t;
+       consumer.m_toSync[key] = new_data;
     }
     /* If an old task is still there, we combine the old task with new task */
     else
     {
-        KeyOpFieldsValuesTuple u = m_toSync[key];
+        KeyOpFieldsValuesTuple existing_data = consumer.m_toSync[key];
 
-        auto tt = kfvFieldsValues(t);
-	auto uu = kfvFieldsValues(u);
+        auto new_values      = kfvFieldsValues(new_data);
+        auto existing_values = kfvFieldsValues(existing_data);
 
 
-        for (auto it = tt.begin(); it != tt.end(); it++)
+        for (auto it = new_values.begin(); it != new_values.end(); it++)
         {
             string field = fvField(*it);
             string value = fvValue(*it);
 
-            auto iu = uu.begin();
-            while (iu != uu.end())
+            auto iu = existing_values.begin();
+            while (iu != existing_values.end())
             {
                 string ofield = fvField(*iu);
                 if (field == ofield)
-                    iu = uu.erase(iu);
+                    iu = existing_values.erase(iu);
                 else
                     iu++;
             }
-            uu.push_back(FieldValueTuple(field, value));
+            existing_values.push_back(FieldValueTuple(field, value));
         }
-        m_toSync[key] = KeyOpFieldsValuesTuple(key, op, uu);
+        consumer.m_toSync[key] = KeyOpFieldsValuesTuple(key, op, existing_values);
     }
 
-    doTask();
+    doTask(consumer);
+    return true;
 }

orchagent/orch.h

@@ -15,28 +15,38 @@ extern "C" {
 using namespace std;
 using namespace swss;
 
+typedef map<string, KeyOpFieldsValuesTuple> SyncMap;
+struct Consumer {
+    Consumer(ConsumerTable* consumer) :m_consumer(consumer)  { }
+    ConsumerTable* m_consumer;
+    /* Store the latest 'golden' status */
+    SyncMap m_toSync;
+};
+typedef std::pair<string, Consumer> ConsumerMapPair;
+typedef map<string, Consumer> ConsumerMap;
+
 class Orch
 {
 public:
     Orch(DBConnector *db, string tableName);
+    Orch(DBConnector *db, vector<string> &tableNames);
     ~Orch();
 
-    inline ConsumerTable *getConsumer() { return m_consumer; }
+    void getSelectables( _out_ std::vector<Selectable*>& selectables);
+    bool is_owned_consumer(ConsumerTable* s)const;
 
-    void execute();
-    virtual void doTask() = 0;
+    bool execute(string tableName);
 
     inline string getOrchName() { return m_name; }
 
+protected:
+    virtual void doTask(_in_ Consumer& consumer_info) = 0;
 private:
     DBConnector *m_db;
-    const string m_name;
+    const string m_name;// TODO: where is this field initialized??
 
 protected:
-    ConsumerTable *m_consumer;
-
-    /* Store the latest 'golden' status */
-    map<string, KeyOpFieldsValuesTuple> m_toSync;
+    ConsumerMap m_consumer_map;
 
 };