Skip to content

Commit dc695fb

Browse files
zhenggen-xulguohan
zhenggen-xu
authored and
lguohan
committed
[orch] change Consumer class to support multiple values for the same key (#1184)
Description The Consumer class is used by the orch objects to deal with the redis consumer tables' popped items. It has m_toSync map to save the tasks. During the operations (more items than the orch objects can handle), the tasks in the map is merged for optimization. However, since it is a map, we would only have one value for each key. This potentially eliminate the necessary actions from Redis, e,g, we have a DEL action and SET action coming out from Redis for some key, we would overwrite the DEL action today in the code and thus not able to delete some objects as we intended. The PR changed the m_toSync from map to multi-map to get multiple values for one key. In this design, we keep maximun two values per key, DEL or SET or DEL+SET. We need strictly keep the order of DEL and SET. It is possible to use map of vectors to fulfill this, we chose multi-map because: 1, It will have less/no changes to different orch classes to iterate the m_toSync 2, The order can be guaranteed. The order of the key-value pairs whose keys compare equivalent is the order of insertion and does not change. (since C++11). See https://en.cppreference.com/w/cpp/container/multimap The PR also refactors the consumer class so vlanmgr.cpp and routeorch.cpp will leverage the Consumer functions instead of operating on the members. It also refactors the UT code (aclorch_ut.cpp) so it removes the redundant code and uses the same code. Google UT tests were added for Consumer Class especially for different cases for addToSync() function. What I did Change the m_toSync in Consumer class to multimap so it could support both DEL and SET Reload Consumer addToSync() and refactor vlanmgr/route-orch and ut code to use it Add google ut for consumer class Why I did it See description. How I verified it Unit tests: Running main() from gtest_main.cc ``` [==========] Running 19 tests from 5 test cases. [----------] Global test environment set-up. [----------] 1 test from AclTest [ RUN ] AclTest.Create_L3_Acl_Table [ OK ] AclTest.Create_L3_Acl_Table (1 ms) [----------] 1 test from AclTest (1 ms total) [----------] 3 tests from AclOrchTest [ RUN ] AclOrchTest.ACL_Creation_and_Destorying [ OK ] AclOrchTest.ACL_Creation_and_Destorying (1000 ms) [ RUN ] AclOrchTest.L3Acl_Matches_Actions [ OK ] AclOrchTest.L3Acl_Matches_Actions (1001 ms) [ RUN ] AclOrchTest.L3V6Acl_Matches_Actions [ OK ] AclOrchTest.L3V6Acl_Matches_Actions (1000 ms) [----------] 3 tests from AclOrchTest (3003 ms total) [----------] 2 tests from PortsOrchTest [ RUN ] PortsOrchTest.PortReadinessColdBoot [ OK ] PortsOrchTest.PortReadinessColdBoot (21 ms) [ RUN ] PortsOrchTest.PortReadinessWarmBoot [ OK ] PortsOrchTest.PortReadinessWarmBoot (13 ms) [----------] 2 tests from PortsOrchTest (34 ms total) [----------] 4 tests from SaiSpy [ RUN ] SaiSpy.CURD [ OK ] SaiSpy.CURD (0 ms) [ RUN ] SaiSpy.Same_Function_Signature_In_Same_API_Table [ OK ] SaiSpy.Same_Function_Signature_In_Same_API_Table (0 ms) [ RUN ] SaiSpy.Same_Function_Signature_In_Different_API_Table [ OK ] SaiSpy.Same_Function_Signature_In_Different_API_Table (0 ms) [ RUN ] SaiSpy.create_switch_and_acl_table [ OK ] SaiSpy.create_switch_and_acl_table (0 ms) [----------] 4 tests from SaiSpy (0 ms total) [----------] 9 tests from ConsumerTest [ RUN ] ConsumerTest.ConsumerAddToSync_Set [ OK ] ConsumerTest.ConsumerAddToSync_Set (1 ms) [ RUN ] ConsumerTest.ConsumerAddToSync_Del [ OK ] ConsumerTest.ConsumerAddToSync_Del (0 ms) [ RUN ] ConsumerTest.ConsumerAddToSync_Set_Del [ OK ] ConsumerTest.ConsumerAddToSync_Set_Del (0 ms) [ RUN ] ConsumerTest.ConsumerAddToSync_Del_Set [ OK ] ConsumerTest.ConsumerAddToSync_Del_Set (130 ms) [ RUN ] ConsumerTest.ConsumerAddToSync_Set_Del_Set_Multi [ OK ] ConsumerTest.ConsumerAddToSync_Set_Del_Set_Multi (204 ms) [ RUN ] ConsumerTest.ConsumerAddToSync_Set_Del_Set_Multi_In_Q [ OK ] ConsumerTest.ConsumerAddToSync_Set_Del_Set_Multi_In_Q (4 ms) [ RUN ] ConsumerTest.ConsumerAddToSync_Del_Set_Setnew [ OK ] ConsumerTest.ConsumerAddToSync_Del_Set_Setnew (0 ms) [ RUN ] ConsumerTest.ConsumerAddToSync_Del_Set_Setnew1 [ OK ] ConsumerTest.ConsumerAddToSync_Del_Set_Setnew1 (0 ms) [ RUN ] ConsumerTest.ConsumerAddToSync_Ind_Set_Del [ OK ] ConsumerTest.ConsumerAddToSync_Ind_Set_Del (0 ms) [----------] 9 tests from ConsumerTest (340 ms total) [----------] Global test environment tear-down [==========] 19 tests from 5 test cases ran. (4344 ms total) [ PASSED ] 19 tests. ``` Signed-off-by: Zhenggen Xu <[email protected]>
1 parent 49ad38f commit dc695fb

File tree

7 files changed

+412
-86
lines changed

7 files changed

+412
-86
lines changed

‎cfgmgr/vlanmgr.cpp

+3-2
Original file line numberDiff line numberDiff line change
@@ -447,8 +447,9 @@ void VlanMgr::processUntaggedVlanMembers(string vlan, const string &members)
447447
vector<FieldValueTuple> fvVector;
448448
FieldValueTuple t("tagging_mode", "untagged");
449449
fvVector.push_back(t);
450-
consumer.m_toSync[member_key] = make_tuple(member_key, SET_COMMAND, fvVector);
451-
SWSS_LOG_DEBUG("%s", (dumpTuple(consumer, consumer.m_toSync[member_key])).c_str());
450+
KeyOpFieldsValuesTuple tuple = make_tuple(member_key, SET_COMMAND, fvVector);
451+
consumer.addToSync(tuple);
452+
SWSS_LOG_DEBUG("%s", (dumpTuple(consumer, tuple)).c_str());
452453
}
453454
/*
454455
* There is pending task from consumer pipe, in this case just skip it.

‎orchagent/orch.cpp

+62-20
Original file line numberDiff line numberDiff line change
@@ -67,36 +67,66 @@ vector<Selectable *> Orch::getSelectables()
6767
return selectables;
6868
}
6969

70-
size_t Consumer::addToSync(std::deque<KeyOpFieldsValuesTuple> &entries)
70+
void Consumer::addToSync(const KeyOpFieldsValuesTuple &entry)
7171
{
7272
SWSS_LOG_ENTER();
7373

74-
/* Nothing popped */
75-
if (entries.empty())
74+
75+
string key = kfvKey(entry);
76+
string op = kfvOp(entry);
77+
78+
/* Record incoming tasks */
79+
if (gSwssRecord)
7680
{
77-
return 0;
81+
Orch::recordTuple(*this, entry);
7882
}
7983

80-
for (auto& entry: entries)
84+
/*
85+
* m_toSync is a multimap which will allow one key with multiple values,
86+
* Also, the order of the key-value pairs whose keys compare equivalent
87+
* is the order of insertion and does not change. (since C++11)
88+
*/
89+
90+
/* If a new task comes we directly put it into getConsumerTable().m_toSync map */
91+
if (m_toSync.find(key) == m_toSync.end())
8192
{
82-
string key = kfvKey(entry);
83-
string op = kfvOp(entry);
93+
m_toSync.emplace(key, entry);
94+
}
8495

85-
/* Record incoming tasks */
86-
if (gSwssRecord)
96+
/* if a DEL task comes, we overwrite the old key */
97+
else if (op == DEL_COMMAND)
98+
{
99+
m_toSync.erase(key);
100+
m_toSync.emplace(key, entry);
101+
}
102+
else
103+
{
104+
/*
105+
* Now we are trying to add the key-value with SET.
106+
* We maintain maximun two values per key.
107+
* In case there is one key-value, it should be DEL or SET
108+
* In case there are two key-value pairs, it should be DEL then SET
109+
* The code logic is following:
110+
* We iterate the values with the key, we skip the value with DEL and then
111+
* check if that was the only one (I,E, the iter pointer now points to end or next key),
112+
* in such case, we insert the key-value with SET.
113+
* If there was a SET already (I,E, the pointer still points to the same key), we combine the kfv.
114+
*/
115+
auto ret = m_toSync.equal_range(key);
116+
auto iter = ret.first;
117+
for (; iter != ret.second; ++iter)
87118
{
88-
Orch::recordTuple(*this, entry);
119+
auto old_op = kfvOp(iter->second);
120+
if (old_op == SET_COMMAND)
121+
break;
89122
}
90-
91-
/* If a new task comes or if a DEL task comes, we directly put it into getConsumerTable().m_toSync map */
92-
if (m_toSync.find(key) == m_toSync.end() || op == DEL_COMMAND)
123+
if (iter == ret.second)
93124
{
94-
m_toSync[key] = entry;
125+
m_toSync.emplace(key, entry);
95126
}
96-
/* If an old task is still there, we combine the old task with new task */
97127
else
98128
{
99-
KeyOpFieldsValuesTuple existing_data = m_toSync[key];
129+
KeyOpFieldsValuesTuple existing_data = iter->second;
100130

101131
auto new_values = kfvFieldsValues(entry);
102132
auto existing_values = kfvFieldsValues(existing_data);
@@ -118,9 +148,21 @@ size_t Consumer::addToSync(std::deque<KeyOpFieldsValuesTuple> &entries)
118148
}
119149
existing_values.push_back(FieldValueTuple(field, value));
120150
}
121-
m_toSync[key] = KeyOpFieldsValuesTuple(key, op, existing_values);
151+
iter->second = KeyOpFieldsValuesTuple(key, op, existing_values);
122152
}
123153
}
154+
155+
}
156+
157+
size_t Consumer::addToSync(const std::deque<KeyOpFieldsValuesTuple> &entries)
158+
{
159+
SWSS_LOG_ENTER();
160+
161+
for (auto& entry: entries)
162+
{
163+
addToSync(entry);
164+
}
165+
124166
return entries.size();
125167
}
126168

@@ -186,7 +228,7 @@ void Consumer::drain()
186228
m_orch->doTask(*this);
187229
}
188230

189-
string Consumer::dumpTuple(KeyOpFieldsValuesTuple &tuple)
231+
string Consumer::dumpTuple(const KeyOpFieldsValuesTuple &tuple)
190232
{
191233
string s = getTableName() + getConsumerTable()->getTableNameSeparator() + kfvKey(tuple)
192234
+ "|" + kfvOp(tuple);
@@ -412,7 +454,7 @@ void Orch::logfileReopen()
412454
}
413455
}
414456

415-
void Orch::recordTuple(Consumer &consumer, KeyOpFieldsValuesTuple &tuple)
457+
void Orch::recordTuple(Consumer &consumer, const KeyOpFieldsValuesTuple &tuple)
416458
{
417459
string s = consumer.dumpTuple(tuple);
418460

@@ -426,7 +468,7 @@ void Orch::recordTuple(Consumer &consumer, KeyOpFieldsValuesTuple &tuple)
426468
}
427469
}
428470

429-
string Orch::dumpTuple(Consumer &consumer, KeyOpFieldsValuesTuple &tuple)
471+
string Orch::dumpTuple(Consumer &consumer, const KeyOpFieldsValuesTuple &tuple)
430472
{
431473
string s = consumer.dumpTuple(tuple);
432474
return s;

‎orchagent/orch.h

+11-6
Original file line numberDiff line numberDiff line change
@@ -57,7 +57,11 @@ typedef std::pair<std::string, sai_object_id_t> object_map_pair;
5757

5858
typedef std::map<std::string, object_map*> type_map;
5959
typedef std::pair<std::string, object_map*> type_map_pair;
60-
typedef std::map<std::string, swss::KeyOpFieldsValuesTuple> SyncMap;
60+
61+
// Use multimap to support multiple OpFieldsValues for the same key (e,g, DEL and SET)
62+
// The order of the key-value pairs whose keys compare equivalent is the order of
63+
// insertion and does not change. (since C++11)
64+
typedef std::multimap<std::string, swss::KeyOpFieldsValuesTuple> SyncMap;
6165

6266
typedef std::pair<std::string, int> table_name_with_pri_t;
6367

@@ -132,7 +136,7 @@ class Consumer : public Executor {
132136
return getConsumerTable()->getDbId();
133137
}
134138

135-
std::string dumpTuple(swss::KeyOpFieldsValuesTuple &tuple);
139+
std::string dumpTuple(const swss::KeyOpFieldsValuesTuple &tuple);
136140
void dumpPendingTasks(std::vector<std::string> &ts);
137141

138142
size_t refillToSync();
@@ -144,9 +148,10 @@ class Consumer : public Executor {
144148
// TODO: hide?
145149
SyncMap m_toSync;
146150

147-
protected:
151+
void addToSync(const swss::KeyOpFieldsValuesTuple &entry);
152+
148153
// Returns: the number of entries added to m_toSync
149-
size_t addToSync(std::deque<swss::KeyOpFieldsValuesTuple> &entries);
154+
size_t addToSync(const std::deque<swss::KeyOpFieldsValuesTuple> &entries);
150155
};
151156

152157
typedef std::map<std::string, std::shared_ptr<Executor>> ConsumerMap;
@@ -194,14 +199,14 @@ class Orch
194199
virtual void doTask(swss::SelectableTimer &timer) { }
195200

196201
/* TODO: refactor recording */
197-
static void recordTuple(Consumer &consumer, swss::KeyOpFieldsValuesTuple &tuple);
202+
static void recordTuple(Consumer &consumer, const swss::KeyOpFieldsValuesTuple &tuple);
198203

199204
void dumpPendingTasks(std::vector<std::string> &ts);
200205
protected:
201206
ConsumerMap m_consumerMap;
202207

203208
static void logfileReopen();
204-
std::string dumpTuple(Consumer &consumer, swss::KeyOpFieldsValuesTuple &tuple);
209+
std::string dumpTuple(Consumer &consumer, const swss::KeyOpFieldsValuesTuple &tuple);
205210
ref_resolve_status resolveFieldRefValue(type_map&, const std::string&, swss::KeyOpFieldsValuesTuple&, sai_object_id_t&);
206211
bool parseIndexRange(const std::string &input, sai_uint32_t &range_low, sai_uint32_t &range_high);
207212
bool parseReference(type_map &type_maps, std::string &ref, std::string &table_name, std::string &object_name);

‎orchagent/routeorch.cpp

+2-2
Original file line numberDiff line numberDiff line change
@@ -134,7 +134,7 @@ std::string RouteOrch::getLinkLocalEui64Addr(void)
134134

135135
uint8_t eui64_interface_id[EUI64_INTF_ID_LEN];
136136
char ipv6_ll_addr[INET6_ADDRSTRLEN] = {0};
137-
137+
138138
/* Link-local IPv6 address autogenerated by kernel with eui64 interface-id
139139
* derived from the MAC address of the host interface.
140140
*/
@@ -406,7 +406,7 @@ void RouteOrch::doTask(Consumer& consumer)
406406
vector<FieldValueTuple> v;
407407
key = vrf + i.first.to_string();
408408
auto x = KeyOpFieldsValuesTuple(key, DEL_COMMAND, v);
409-
consumer.m_toSync[key] = x;
409+
consumer.addToSync(x);
410410
}
411411
}
412412
m_resync = true;

‎tests/mock_tests/Makefile.am

+1
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@ LDADD_GTEST = -L/usr/src/gtest
2323
tests_SOURCES = aclorch_ut.cpp \
2424
portsorch_ut.cpp \
2525
saispy_ut.cpp \
26+
consumer_ut.cpp \
2627
ut_saihelper.cpp \
2728
mock_orchagent_main.cpp \
2829
mock_dbconnector.cpp \

‎tests/mock_tests/aclorch_ut.cpp

+5-56
Original file line numberDiff line numberDiff line change
@@ -23,54 +23,6 @@ namespace aclorch_test
2323
{
2424
using namespace std;
2525

26-
size_t consumerAddToSync(Consumer *consumer, const deque<KeyOpFieldsValuesTuple> &entries)
27-
{
28-
/* Nothing popped */
29-
if (entries.empty())
30-
{
31-
return 0;
32-
}
33-
34-
for (auto &entry : entries)
35-
{
36-
string key = kfvKey(entry);
37-
string op = kfvOp(entry);
38-
39-
/* If a new task comes or if a DEL task comes, we directly put it into getConsumerTable().m_toSync map */
40-
if (consumer->m_toSync.find(key) == consumer->m_toSync.end() || op == DEL_COMMAND)
41-
{
42-
consumer->m_toSync[key] = entry;
43-
}
44-
/* If an old task is still there, we combine the old task with new task */
45-
else
46-
{
47-
KeyOpFieldsValuesTuple existing_data = consumer->m_toSync[key];
48-
49-
auto new_values = kfvFieldsValues(entry);
50-
auto existing_values = kfvFieldsValues(existing_data);
51-
52-
for (auto it : new_values)
53-
{
54-
string field = fvField(it);
55-
string value = fvValue(it);
56-
57-
auto iu = existing_values.begin();
58-
while (iu != existing_values.end())
59-
{
60-
string ofield = fvField(*iu);
61-
if (field == ofield)
62-
iu = existing_values.erase(iu);
63-
else
64-
iu++;
65-
}
66-
existing_values.push_back(FieldValueTuple(field, value));
67-
}
68-
consumer->m_toSync[key] = KeyOpFieldsValuesTuple(key, op, existing_values);
69-
}
70-
}
71-
return entries.size();
72-
}
73-
7426
struct AclTestBase : public ::testing::Test
7527
{
7628
vector<int32_t *> m_s32list_pool;
@@ -199,8 +151,7 @@ namespace aclorch_test
199151
auto consumer = unique_ptr<Consumer>(new Consumer(
200152
new swss::ConsumerStateTable(config_db, CFG_ACL_TABLE_TABLE_NAME, 1, 1), m_aclOrch, CFG_ACL_TABLE_TABLE_NAME));
201153

202-
consumerAddToSync(consumer.get(), entries);
203-
154+
consumer->addToSync(entries);
204155
static_cast<Orch *>(m_aclOrch)->doTask(*consumer);
205156
}
206157

@@ -209,8 +160,7 @@ namespace aclorch_test
209160
auto consumer = unique_ptr<Consumer>(new Consumer(
210161
new swss::ConsumerStateTable(config_db, CFG_ACL_RULE_TABLE_NAME, 1, 1), m_aclOrch, CFG_ACL_RULE_TABLE_NAME));
211162

212-
consumerAddToSync(consumer.get(), entries);
213-
163+
consumer->addToSync(entries);
214164
static_cast<Orch *>(m_aclOrch)->doTask(*consumer);
215165
}
216166

@@ -381,8 +331,7 @@ namespace aclorch_test
381331
auto consumer = unique_ptr<Consumer>(new Consumer(
382332
new swss::ConsumerStateTable(m_app_db.get(), APP_PORT_TABLE_NAME, 1, 1), gPortsOrch, APP_PORT_TABLE_NAME));
383333

384-
consumerAddToSync(consumer.get(), { { "PortInitDone", EMPTY_PREFIX, { { "", "" } } } });
385-
334+
consumer->addToSync({ { "PortInitDone", EMPTY_PREFIX, { { "", "" } } } });
386335
static_cast<Orch *>(gPortsOrch)->doTask(*consumer.get());
387336
}
388337

@@ -628,7 +577,7 @@ namespace aclorch_test
628577
// consistency validation with CRM
629578
bool validateResourceCountWithCrm(const AclOrch *aclOrch, CrmOrch *crmOrch)
630579
{
631-
// Verify ACL Tables
580+
// Verify ACL Tables
632581
auto const &resourceMap = Portal::CrmOrchInternal::getResourceMap(crmOrch);
633582
uint32_t crm_acl_table_cnt = 0;
634583
for (auto const &kv : resourceMap.at(CrmResourceType::CRM_ACL_TABLE).countersMap)
@@ -642,7 +591,7 @@ namespace aclorch_test
642591
<< ") and AclOrch " << Portal::AclOrchInternal::getAclTables(aclOrch).size();
643592
return false;
644593
}
645-
594+
646595

647596
// Verify ACL Rules
648597
//

0 commit comments

Comments
 (0)