Skip to content

Refactor Selectables. Add support for selectable priorities #192

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

Merged
lguohan merged 16 commits into from
Apr 16, 2018
Merged

Refactor Selectables. Add support for selectable priorities #192

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
16 commits
Select commit Hold shift + click to select a range
63bf401
Refactor select and selectable. Now they support priority
pavel-shirshov Apr 3, 2018
cfee528
Remove fd parameter from Select.select(). This parameter was never used
pavel-shirshov Apr 3, 2018
bac891b
Use references as loop variables
pavel-shirshov Apr 3, 2018
fb61314
Make select fair for selectables.
pavel-shirshov Apr 11, 2018
8e328aa
Revert "Use references as loop variables"
pavel-shirshov Apr 11, 2018
5b76223
Revert performance changes
pavel-shirshov Apr 11, 2018
4d3bcea
Address comments
pavel-shirshov Apr 11, 2018
e835cac
Make the comparator comment clearer
pavel-shirshov Apr 11, 2018
28dcdad
Merge branch 'master' into pavelsh/pri
pavel-shirshov Apr 12, 2018
e428d9c
Use right EINTR error code for netlink function
pavel-shirshov Apr 12, 2018
88dd5fa
Merge branch 'master' into pavelsh/pri
pavel-shirshov Apr 12, 2018
842b5b0
Move Selectable comparator to the Select class as a private functiona…
pavel-shirshov Apr 13, 2018
7bfc1ef
Merge branch 'master' into pavelsh/pri
pavel-shirshov Apr 13, 2018
4d711bb
Hide m_last_used_time from all classes except Select
pavel-shirshov Apr 13, 2018
21b3445
Rename some class members
pavel-shirshov Apr 13, 2018
7359660
Make more consistent naming for LastUsedTime of selectable
pavel-shirshov Apr 13, 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
4 changes: 2 additions & 2 deletions common/consumerstatetable.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,8 +11,8 @@

namespace swss {

ConsumerStateTable::ConsumerStateTable(DBConnector *db, std::string tableName, int popBatchSize)
: ConsumerTableBase(db, tableName, popBatchSize)
ConsumerStateTable::ConsumerStateTable(DBConnector *db, std::string tableName, int popBatchSize, int pri)
: ConsumerTableBase(db, tableName, popBatchSize, pri)
, TableName_KeySet(tableName)
{
for (;;)
Expand Down
2 changes: 1 addition & 1 deletion common/consumerstatetable.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@ namespace swss {
class ConsumerStateTable : public ConsumerTableBase, public TableName_KeySet
{
public:
ConsumerStateTable(DBConnector *db, std::string tableName, int popBatchSize = DEFAULT_POP_BATCH_SIZE);
ConsumerStateTable(DBConnector *db, std::string tableName, int popBatchSize = DEFAULT_POP_BATCH_SIZE, int pri = 0);

/* Get multiple pop elements */
void pops(std::deque<KeyOpFieldsValuesTuple> &vkco, std::string prefix = EMPTY_PREFIX);
Expand Down
4 changes: 2 additions & 2 deletions common/consumertable.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,8 +14,8 @@ using namespace std;

namespace swss {

ConsumerTable::ConsumerTable(DBConnector *db, string tableName, int popBatchSize)
: ConsumerTableBase(db, tableName, popBatchSize)
ConsumerTable::ConsumerTable(DBConnector *db, string tableName, int popBatchSize, int pri)
: ConsumerTableBase(db, tableName, popBatchSize, pri)
, TableName_KeyValueOpQueues(tableName)
{
for (;;)
Expand Down
2 changes: 1 addition & 1 deletion common/consumertable.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@ namespace swss {
class ConsumerTable : public ConsumerTableBase, public TableName_KeyValueOpQueues
{
public:
ConsumerTable(DBConnector *db, std::string tableName, int popBatchSize = DEFAULT_POP_BATCH_SIZE);
ConsumerTable(DBConnector *db, std::string tableName, int popBatchSize = DEFAULT_POP_BATCH_SIZE, int pri = 0);

/* Get multiple pop elements */
void pops(std::deque<KeyOpFieldsValuesTuple> &vkco, std::string prefix = EMPTY_PREFIX);
Expand Down
4 changes: 2 additions & 2 deletions common/consumertablebase.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,8 +2,8 @@

namespace swss {

ConsumerTableBase::ConsumerTableBase(DBConnector *db, std::string tableName, int popBatchSize):
TableConsumable(tableName),
ConsumerTableBase::ConsumerTableBase(DBConnector *db, std::string tableName, int popBatchSize, int pri):
TableConsumable(tableName, pri),
RedisTransactioner(db),
POP_BATCH_SIZE(popBatchSize)
{
Expand Down
2 changes: 1 addition & 1 deletion common/consumertablebase.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@ class ConsumerTableBase: public TableConsumable, public RedisTransactioner
public:
const int POP_BATCH_SIZE;

ConsumerTableBase(DBConnector *db, std::string tableName, int popBatchSize = DEFAULT_POP_BATCH_SIZE);
ConsumerTableBase(DBConnector *db, std::string tableName, int popBatchSize = DEFAULT_POP_BATCH_SIZE, int pri = 0);

virtual ~ConsumerTableBase();

Expand Down
2 changes: 1 addition & 1 deletion common/ipaddress.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@ class IpAddress
{
public:
IpAddress() {}
IpAddress(const ip_addr_t &ip) { m_ip = ip; }
IpAddress(const ip_addr_t &ip) : m_ip(ip) {}
Copy link
Copy Markdown
Contributor

@qiluo-msft qiluo-msft Apr 3, 2018 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

m_ip [](start = 37, length = 4)

Could you separate simple refactoring code like this into a smaller PR? so we could focus on more difficult one. #Closed

Copy link
Copy Markdown
Contributor Author

@pavel-shirshov pavel-shirshov Apr 3, 2018 •
edited by qiluo-msft
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I can, but it will take more time to extract all the changes and revert them back. Can you please consider to check them? They're tiny and follows just a few easy patterns:

func(object obj) -> func(const object& obj)

,

for(auto obj: object) -> for (const auto& obj: object)

'''
Class1::Class1(int var1)
{
m_var_1 = var1;
}
to
Class1::Class1(int var1) : m_var_1(var1) {}

 #Closed

Copy link
Copy Markdown
Contributor

@qiluo-msft qiluo-msft Apr 5, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I am thinking 'selectable improvement' as an experiment feature (not mature now). And other simple one are definitely good contribution to existing codebase. Considering the possibility of future reverting and/or conflicting, it's good to separate. Large amount of files in a PR always harm.

In reply to: 178991122 [](ancestors = 178991122)

Copy link
Copy Markdown
Contributor Author

@pavel-shirshov pavel-shirshov Apr 11, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I've removed the other changes. I kept only one bug with the missed va_end()

IpAddress(uint32_t ip);
IpAddress(const std::string &ipStr);

Expand Down
4 changes: 2 additions & 2 deletions common/ipaddresses.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,7 @@ using namespace swss;
IpAddresses::IpAddresses(const string &ipsStr)
{
auto ips = tokenize(ipsStr, IP_DELIMITER);
for (auto ip : ips)
for (const auto& ip : ips)
m_ips.insert(ip);
}

Expand Down Expand Up @@ -39,7 +39,7 @@ bool IpAddresses::contains(const IpAddress &ip) const

bool IpAddresses::contains(const IpAddresses &ips) const
{
for (auto ip : ips.getIpAddresses())
for (const auto& ip : ips.getIpAddresses())
{
if (!contains(ip))
{
Expand Down
5 changes: 2 additions & 3 deletions common/ipprefix.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,10 +45,9 @@ IpPrefix::IpPrefix(
}
}

IpPrefix::IpPrefix(uint32_t ipPrefix, int mask)
IpPrefix::IpPrefix(uint32_t ipPrefix, int mask) : m_ip(ipPrefix),
m_mask(mask)
{
m_ip = IpAddress(ipPrefix);
m_mask = mask;
if (!isValid())
{
throw std::invalid_argument("Invalid IpPrefix from prefix and mask");
Expand Down
2 changes: 1 addition & 1 deletion common/json.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@ string JSon::buildJson(const vector<FieldValueTuple> &fv)
nlohmann::json j = nlohmann::json::array();

// we use array to save order
for (auto &i : fv)
for (const auto& i : fv)
{
j.push_back(fvField(i));
j.push_back(fvValue(i));
Expand Down
9 changes: 4 additions & 5 deletions common/logger.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -70,7 +70,7 @@ void Logger::swssOutputNotify(std::string component, std::string outputStr)
}
}

void Logger::linkToDbWithOutput(const std::string dbName, const PriorityChangeNotify& prioNotify, const std::string& defPrio, const OutputChangeNotify& outputNotify, const std::string& defOutput)
void Logger::linkToDbWithOutput(const std::string& dbName, const PriorityChangeNotify& prioNotify, const std::string& defPrio, const OutputChangeNotify& outputNotify, const std::string& defOutput)
{
auto& logger = getInstance();

Expand Down Expand Up @@ -122,12 +122,12 @@ void Logger::linkToDbWithOutput(const std::string dbName, const PriorityChangeNo
outputNotify(key, output);
}

void Logger::linkToDb(const std::string dbName, const PriorityChangeNotify& prioNotify, const std::string& defPrio)
void Logger::linkToDb(const std::string& dbName, const PriorityChangeNotify& prioNotify, const std::string& defPrio)
{
linkToDbWithOutput(dbName, prioNotify, defPrio, swssOutputNotify, "SYSLOG");
}

void Logger::linkToDbNative(const std::string dbName)
void Logger::linkToDbNative(const std::string& dbName)
{
auto& logger = getInstance();

Expand Down Expand Up @@ -166,10 +166,9 @@ Logger::Priority Logger::getMinPrio()

while(true)
{
int fd = 0;
Selectable *selectable = nullptr;

int ret = select.select(&selectable, &fd);
int ret = select.select(&selectable);

if (ret == Select::ERROR)
{
Expand Down
6 changes: 3 additions & 3 deletions common/logger.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -55,10 +55,10 @@ class Logger
static Logger &getInstance();
static void setMinPrio(Priority prio);
static Priority getMinPrio();
static void linkToDbWithOutput(const std::string dbName, const PriorityChangeNotify& prioNotify, const std::string& defPrio, const OutputChangeNotify& outputNotify, const std::string& defOutput);
static void linkToDb(const std::string dbName, const PriorityChangeNotify& notify, const std::string& defPrio);
static void linkToDbWithOutput(const std::string& dbName, const PriorityChangeNotify& prioNotify, const std::string& defPrio, const OutputChangeNotify& outputNotify, const std::string& defOutput);
static void linkToDb(const std::string& dbName, const PriorityChangeNotify& notify, const std::string& defPrio);
// Must be called after all linkToDb to start select from DB
static void linkToDbNative(const std::string dbName);
static void linkToDbNative(const std::string& dbName);
void write(Priority prio, const char *fmt, ...)
#ifdef __GNUC__
__attribute__ ((format (printf, 3, 4)))
Expand Down
25 changes: 11 additions & 14 deletions common/netlink.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,8 +9,8 @@
using namespace swss;
using namespace std;

NetLink::NetLink() :
m_socket(NULL)
NetLink::NetLink(int pri) :
Selectable(pri), m_socket(NULL)
{
m_socket = nl_socket_alloc();
if (!m_socket)
Expand Down Expand Up @@ -71,24 +71,21 @@ void NetLink::dumpRequest(int rtmGetCommand)
}
}

void NetLink::addFd(fd_set *fd)
int NetLink::getFd()
{
FD_SET(nl_socket_get_fd(m_socket), fd);
return nl_socket_get_fd(m_socket);
}

bool NetLink::isMe(fd_set *fd)
void NetLink::readData()
{
return FD_ISSET(nl_socket_get_fd(m_socket), fd);
}
int err;

int NetLink::readCache()
{
return NODATA;
}
do
{
err = nl_recvmsgs_default(m_socket);
}
while(err == -EINTR); // Retry if the process was interrupted by a signal

void NetLink::readMe()
{
int err = nl_recvmsgs_default(m_socket);
if (err < 0)
{
if (err == -NLE_NOMEM)
Expand Down
8 changes: 3 additions & 5 deletions common/netlink.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,16 +9,14 @@ namespace swss {

class NetLink : public Selectable {
public:
NetLink();
NetLink(int pri = 0);
virtual ~NetLink();

void registerGroup(int rtnlGroup);
void dumpRequest(int rtmGetCommand);

virtual void addFd(fd_set *fd);
virtual bool isMe(fd_set *fd);
virtual int readCache();
virtual void readMe();
int getFd() override;
void readData() override;

private:
static int onNetlinkMsg(struct nl_msg *msg, void *arg);
Expand Down
98 changes: 45 additions & 53 deletions common/notificationconsumer.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,8 @@
#define REDIS_PUBLISH_MESSAGE_INDEX (2)
#define REDIS_PUBLISH_MESSAGE_ELEMNTS (3)

swss::NotificationConsumer::NotificationConsumer(swss::DBConnector *db, std::string channel):
swss::NotificationConsumer::NotificationConsumer(swss::DBConnector *db, std::string channel, int pri):
Selectable(pri),
m_db(db),
m_subscribe(NULL),
m_channel(channel)
Expand Down Expand Up @@ -56,11 +57,51 @@ void swss::NotificationConsumer::subscribe()
SWSS_LOG_INFO("subscribed to %s", m_channel.c_str());
}

void swss::NotificationConsumer::addFd(fd_set *fd)
int swss::NotificationConsumer::getFd()
{
return m_subscribe->getContext()->fd;
}

void swss::NotificationConsumer::readData()
{
SWSS_LOG_ENTER();

FD_SET(m_subscribe->getContext()->fd, fd);
redisReply *reply = nullptr;

if (redisGetReply(m_subscribe->getContext(), reinterpret_cast<void**>(&reply)) != REDIS_OK)
{
SWSS_LOG_ERROR("failed to read redis reply on channel %s", m_channel.c_str());

throw std::runtime_error("Unable to read redis reply");
}
else
{
RedisReply r(reply);
processReply(reply);
}

reply = nullptr;
int status;
do
{
status = redisGetReplyFromReader(m_subscribe->getContext(), reinterpret_cast<void**>(&reply));
if(reply != nullptr && status == REDIS_OK)
{
RedisReply r(reply);
processReply(reply);
}
}
while(reply != nullptr && status == REDIS_OK);

if (status != REDIS_OK)
{
throw std::runtime_error("Unable to read redis reply");
}
}

bool swss::NotificationConsumer::hasCachedData()
{
return m_queue.size() > 1;
}

void swss::NotificationConsumer::processReply(redisReply *reply)
Expand Down Expand Up @@ -91,60 +132,11 @@ void swss::NotificationConsumer::processReply(redisReply *reply)
m_queue.push(msg);
}

int swss::NotificationConsumer::readCache()
{
SWSS_LOG_ENTER();

if (m_queue.size() > 0)
{
return Selectable::DATA;
}

redisReply *reply = NULL;

if (redisGetReplyFromReader(m_subscribe->getContext(), (void**)&reply) != REDIS_OK)
{
SWSS_LOG_ERROR("failed to read redis reply on channel %s", m_channel.c_str());

return Selectable::ERROR;
}
else if (reply != NULL)
{
RedisReply r(reply);
processReply(reply);
return Selectable::DATA;
}

return Selectable::NODATA;
}

void swss::NotificationConsumer::readMe()
{
SWSS_LOG_ENTER();

redisReply *reply = NULL;

if (redisGetReply(m_subscribe->getContext(), (void**)&reply) != REDIS_OK)
{
SWSS_LOG_ERROR("failed to read redis reply on channel %s", m_channel.c_str());

throw std::runtime_error("Unable to read redis reply");
}

RedisReply r(reply);
processReply(reply);
}

bool swss::NotificationConsumer::isMe(fd_set *fd)
{
return FD_ISSET(m_subscribe->getContext()->fd, fd);
}

void swss::NotificationConsumer::pop(std::string &op, std::string &data, std::vector<FieldValueTuple> &values)
{
SWSS_LOG_ENTER();

if (m_queue.size() == 0)
if (m_queue.empty())
{
SWSS_LOG_ERROR("notification queue is empty, can't pop");
throw std::runtime_error("notification queue is empty, can't pop");
Expand Down
9 changes: 4 additions & 5 deletions common/notificationconsumer.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,16 +19,15 @@ namespace swss {
class NotificationConsumer : public Selectable
{
public:
NotificationConsumer(swss::DBConnector *db, std::string channel);
NotificationConsumer(swss::DBConnector *db, std::string channel, int pri = 100);

void pop(std::string &op, std::string &data, std::vector<FieldValueTuple> &values);

virtual ~NotificationConsumer();

virtual void addFd(fd_set *fd);
virtual bool isMe(fd_set *fd);
virtual int readCache();
virtual void readMe();
int getFd() override;
void readData() override;
bool hasCachedData() override;

private:

Expand Down
2 changes: 1 addition & 1 deletion common/producerstatetable.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -69,7 +69,7 @@ void ProducerStateTable::set(string key, vector<FieldValueTuple> &values,

args.push_back("G");
args.push_back(key);
for (auto& iv: values)
for (const auto& iv: values)
{
args.push_back(fvField(iv));
args.push_back(fvValue(iv));
Expand Down
2 changes: 1 addition & 1 deletion common/producertable.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -89,7 +89,7 @@ void ProducerTable::set(string key, vector<FieldValueTuple> &values, string op,
json j;
string json_key = getKeyName(key);
j[json_key] = json::object();
for (auto it : values)
for (const auto& it : values)
j[json_key][fvField(it)] = fvValue(it);
j["OP"] = op;
m_dumpFile << j.dump(4);
Expand Down
Loading