Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Improve ConsumerStateTable: del after consumed, and let consumer writes the final table #204

Merged
merged 5 commits into from
Jul 27, 2018
Merged

Improve ConsumerStateTable: del after consumed, and let consumer writes the final table #204

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
f23ddd5
Add test assertion on state queue empty
qiluo-msft May 8, 2018
5dcd3f0
Improve ConsumerStateTable: del after consumed
qiluo-msft May 23, 2018
48b51a6
Fix Producer/ConsumerStateTable: let consumer writes the final table
qiluo-msft May 24, 2018
b073815
Fix fieldvalues loop
qiluo-msft Jul 14, 2018
2d9759f
Fix DEL, add tests
qiluo-msft Jul 14, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 12 additions & 2 deletions common/consumer_state_table_pops.lua
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,9 +1,19 @@
redis.replicate_commands()
local ret = {}
local tablename = KEYS[2]
local stateprefix = ARGV[2]
local keys = redis.call('SPOP', KEYS[1], ARGV[1])
local n = table.getn(keys)
for i = 1, n do
local key = keys[i]
local values = redis.call('HGETALL', KEYS[2] .. key)
table.insert(ret, {key, values})
local fieldvalues = redis.call('HGETALL', stateprefix..tablename..key)
table.insert(ret, {key, fieldvalues})
for i = 1, #fieldvalues, 2 do
redis.call('HSET', tablename..key, fieldvalues[i], fieldvalues[i + 1])
end
redis.call('DEL', stateprefix..tablename..key)
if #fieldvalues == 0 then
redis.call('DEL', tablename..key)
end
end
return ret
6 changes: 3 additions & 3 deletions common/consumerstatetable.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,11 +38,12 @@ void ConsumerStateTable::pops(std::deque<KeyOpFieldsValuesTuple> &vkco, const st

RedisCommand command;
command.format(
"EVALSHA %s 2 %s %s: %d ''",
"EVALSHA %s 2 %s %s: %d %s",
sha.c_str(),
getKeySetName().c_str(),
getTableName().c_str(),
POP_BATCH_SIZE);
POP_BATCH_SIZE,
getStateHashPrefix().c_str());

RedisReply r(m_db, command);
auto ctx0 = r.getContext();
Expand All @@ -64,7 +65,6 @@ void ConsumerStateTable::pops(std::deque<KeyOpFieldsValuesTuple> &vkco, const st
assert(values.empty());

auto& ctx = ctx0->element[ie];
assert(ctx->elements == 2);
assert(ctx->element[0]->type == REDIS_REPLY_STRING);
std::string key = ctx->element[0]->str;
kfvKey(kco) = key;
Expand Down
4 changes: 2 additions & 2 deletions common/producerstatetable.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -65,7 +65,7 @@ void ProducerStateTable::set(const string &key, const vector<FieldValueTuple> &v
args.emplace_back(getChannelName());
args.emplace_back(getKeySetName());

args.insert(args.end(), values.size(), getKeyName(key));
args.insert(args.end(), values.size(), getStateHashPrefix() + getKeyName(key));

args.emplace_back("G");
args.emplace_back(key);
Expand Down Expand Up @@ -98,7 +98,7 @@ void ProducerStateTable::del(const string &key, const string &op /*= DEL_COMMAND
args.emplace_back("3");
args.emplace_back(getChannelName());
args.emplace_back(getKeySetName());
args.emplace_back(getKeyName(key));
args.emplace_back(getStateHashPrefix() + getKeyName(key));
args.emplace_back("G");
args.emplace_back(key);
args.emplace_back("''");
Expand Down
1 change: 1 addition & 0 deletions common/table.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -192,6 +192,7 @@ class TableName_KeySet {
}

std::string getKeySetName() const { return m_key; }
std::string getStateHashPrefix() const { return "_"; }
};

}
Expand Down
53 changes: 50 additions & 3 deletions tests/redis_state_ut.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@
#include "common/table.h"
#include "common/producerstatetable.h"
#include "common/consumerstatetable.h"
#include "common/redisclient.h"

using namespace std;
using namespace swss;
Expand Down Expand Up @@ -103,6 +104,7 @@ static void consumerWorker(int index)
{
string tableName = "UT_REDIS_THREAD_" + to_string(index);
DBConnector db(TEST_DB, "localhost", 6379, 0);
RedisClient redisClient(&db);
ConsumerStateTable c(&db, tableName);
Select cs;
Selectable *selectcs;
Expand All @@ -119,9 +121,17 @@ static void consumerWorker(int index)
{
numberOfKeysSet++;
validateFields(kfvKey(kco), kfvFieldsValues(kco));

for (auto fv : kfvFieldsValues(kco))
{
string val = *redisClient.hget(tableName + ":" + kfvKey(kco), fvField(fv));
EXPECT_EQ(val, fvValue(fv));
}
} else if (kfvOp(kco) == "DEL")
{
numberOfKeyDeleted++;
auto keys = redisClient.keys(tableName + ":" + kfvKey(kco));
EXPECT_EQ(0UL, keys.size());
}

if ((i++ % 100) == 0)
Expand Down Expand Up @@ -215,6 +225,13 @@ TEST(ConsumerStateTable, double_set)
int ret = cs.select(&selectcs, 1000);
EXPECT_EQ(ret, Select::TIMEOUT);
}

/* State Queue should be empty */
RedisCommand keys;
keys.format("KEYS %s*", (c.getStateHashPrefix() + tableName).c_str());
RedisReply r(&db, keys, REDIS_REPLY_ARRAY);
auto qlen = r.getContext()->elements;
EXPECT_EQ(qlen, 0U);
}

TEST(ConsumerStateTable, set_del)
Expand Down Expand Up @@ -267,6 +284,13 @@ TEST(ConsumerStateTable, set_del)
int ret = cs.select(&selectcs, 1000);
EXPECT_EQ(ret, Select::TIMEOUT);
}

/* State Queue should be empty */
RedisCommand keys;
keys.format("KEYS %s*", (p.getStateHashPrefix() + tableName).c_str());
RedisReply r(&db, keys, REDIS_REPLY_ARRAY);
auto qlen = r.getContext()->elements;
EXPECT_EQ(qlen, 0U);
}

TEST(ConsumerStateTable, set_del_set)
Expand Down Expand Up @@ -341,6 +365,13 @@ TEST(ConsumerStateTable, set_del_set)
int ret = cs.select(&selectcs, 1000);
EXPECT_EQ(ret, Select::TIMEOUT);
}

/* State Queue should be empty */
RedisCommand keys;
keys.format("KEYS %s*", (c.getStateHashPrefix() + tableName).c_str());
RedisReply r(&db, keys, REDIS_REPLY_ARRAY);
auto qlen = r.getContext()->elements;
EXPECT_EQ(qlen, 0U);
}

TEST(ConsumerStateTable, singlethread)
Expand Down Expand Up @@ -415,6 +446,13 @@ TEST(ConsumerStateTable, singlethread)

cout << "Done. Waiting for all job to finish " << NUMBER_OF_OPS << " jobs." << endl;

/* State Queue should be empty */
RedisCommand keys;
keys.format("KEYS %s*", (c.getStateHashPrefix() + tableName).c_str());
RedisReply r(&db, keys, REDIS_REPLY_ARRAY);
auto qlen = r.getContext()->elements;
EXPECT_EQ(qlen, 0U);

cout << endl << "Done." << endl;
}

Expand Down Expand Up @@ -442,13 +480,15 @@ TEST(ConsumerStateTable, test)
consumerThreads[i]->join();
delete consumerThreads[i];
}

cout << endl << "Done." << endl;
}

TEST(ConsumerStateTable, multitable)
{
DBConnector db(TEST_DB, "localhost", 6379, 0);
ConsumerStateTable *consumers[NUMBER_OF_THREADS];
vector<string> tablenames(NUMBER_OF_THREADS);
thread *producerThreads[NUMBER_OF_THREADS];
KeyOpFieldsValuesTuple kco;
Select cs;
Expand All @@ -463,8 +503,8 @@ TEST(ConsumerStateTable, multitable)
/* Starting the consumer before the producer */
for (i = 0; i < NUMBER_OF_THREADS; i++)
{
consumers[i] = new ConsumerStateTable(&db, string("UT_REDIS_THREAD_") +
to_string(i));
tablenames[i] = string("UT_REDIS_THREAD_") + to_string(i);
consumers[i] = new ConsumerStateTable(&db, tablenames[i]);
producerThreads[i] = new thread(producerWorker, i);
}

Expand Down Expand Up @@ -504,6 +544,13 @@ TEST(ConsumerStateTable, multitable)
delete producerThreads[i];
}


/* State Queue should be empty */
RedisCommand keys;
keys.format("KEYS %s*", (consumers[0]->getStateHashPrefix() + tablenames[0]).c_str());
RedisReply r(&db, keys, REDIS_REPLY_ARRAY);
auto qlen = r.getContext()->elements;
EXPECT_EQ(qlen, 0U);

cout << endl << "Done." << endl;
}