sonic orchagent线程的调度最小单位是Consumer。Consumer是在epoll事件Selectable的基础上的进一步封装,每一次发生epoll事件会触发orchagent进行一次调度。orch是资源的集合,一个orch可以包含多个Consumer,比如acl orch会监听多个redistable。
class Executor
// Design assumption
// 1. one Orch can have one or more Executor
// 2. one Executor must belong to one and only one Orch
// 3. Executor will hold an pointer to new-ed selectable, and delete it during dtor
// 设计假设:
// 1. 一个orch可以拥有一个或者多个Executor
// 2. 一个Executor必须属于一个orch而且仅仅属于一个orch
// 3. Executor有一个指针指向一个new出来的Selectable结构,必须在析构函数中将其删除,否则会泄漏
class Executor : public Selectable
{
public:
Executor(Selectable *selectable, Orch *orch)
: m_selectable(selectable)
, m_orch(orch)
{
}
virtual ~Executor() { delete m_selectable; }
// Decorating Selectable
int getFd() override { return m_selectable->getFd(); }
void readData() override { m_selectable->readData(); }
bool hasCachedData() override { return m_selectable->hasCachedData(); }
bool initializedWithData() override { return m_selectable->initializedWithData(); }
void updateAfterRead() override { m_selectable->updateAfterRead(); }
Orch * getorch() { return m_orch; }
// Disable copying
Executor(const Executor&) = delete;
Executor& operator=(const Executor&) = delete;
// Execute on event happening
// execute执行事件,drain是一个辅助函数
virtual void execute() { }
virtual void drain() { }
protected:
Selectable *m_selectable;//指向new出来的Selectable
Orch *m_orch;//指向一个orch
// Get the underlying selectable 获取指向的Selectable
Selectable *getSelectable() const { return m_selectable; }
};
class Executor只是一个中间的派生类,orch直接使用的是class Consumer和class ExecutableTimer。
class Consumer
消费者类一般用于处理app_db的订阅事件,对于asic_db一般是处理syncd的应答事件。
typedef std::pair<std::string, std::string> FieldValueTuple;
#define fvField std::get<0>
#define fvValue std::get<1>
typedef std::tuple<std::string, std::string, std::vector<FieldValueTuple> > KeyOpFieldsValuesTuple;
#define kfvKey std::get<0>
#define kfvOp std::get<1>
#define kfvFieldsValues std::get<2>
typedef map<string, KeyOpFieldsValuesTuple> SyncMap;
class Consumer : public Executor {
public:
Consumer(TableConsumable *select, Orch *orch)
: Executor(select, orch)
{
}
TableConsumable *getConsumerTable() const
{
return static_cast<TableConsumable *>(getSelectable());
}
string getTableName() const
{
return getConsumerTable()->getTableName();
}
// 事物执行
void execute();
void drain();
/* Store the latest 'golden' status */
// TODO: hide?
SyncMap m_toSync;
};
void Consumer::execute()
epoll事件触发后,需要调用该函数从数据库中读取出指定key的内容,将其加工后存放在m_toSync中,供后续处理。
void Consumer::execute()
{
SWSS_LOG_ENTER();
std::deque<KeyOpFieldsValuesTuple> entries;
//调用pops函数,从redis数据库中读取数据,返回KeyOpFieldsValuesTuple结构
getConsumerTable()->pops(entries);
/* Nothing popped */
if (entries.empty())
{
return;
}
// 遍历每一个事件
for (auto& entry: entries)
{
string key = kfvKey(entry);
string op = kfvOp(entry);
/* Record incoming tasks 记录事件 */
if (gSwssRecord)
{
Orch::recordTuple(*this, entry);
}
/* If a new task comes or if a DEL task comes, we directly put it into getConsumerTable().m_toSync map */
// 在这里进行一次合并,对于删除事件,直接覆盖
if (m_toSync.find(key) == m_toSync.end() || op == DEL_COMMAND)
{
m_toSync[key] = entry;
}
/* If an old task is still there, we combine the old task with new task */
/* */
else
{
KeyOpFieldsValuesTuple existing_data = m_toSync[key];
auto new_values = kfvFieldsValues(entry);
auto existing_values = kfvFieldsValues(existing_data);
//遍历每一个新的值
for (auto it : new_values)
{
string field = fvField(it);
string value = fvValue(it);
auto iu = existing_values.begin();
while (iu != existing_values.end())//遍历每一个旧的值
{
string ofield = fvField(*iu);
if (field == ofield)//相同的域,将老的值覆盖,这里应该跳出while,代码效率较差
iu = existing_values.erase(iu);
else
iu++;
}
/* 将新的值添加进去 */
existing_values.push_back(FieldValueTuple(field, value));
}
m_toSync[key] = KeyOpFieldsValuesTuple(key, op, existing_values);
}
}
//执行所有整理好的任务。
drain();
}
假设有一个task的键值对如下:
key=test;op=set;value={
A:a,
B:b,
C:c,
}
第一次触发任务是在APP_DB中写入了:
key=test;op=set;value={
A:a,
B:b
}
加入orchagent只是将该任务读取到了m_toSync中,由于某种原因没有执行完该任务,依然驻留在m_toSync中。第二次写入了:
key=test;op=set;value={
B:b1,
C:c
}
那么经过execute函数后m_toSync中将会是:
key=test;op=set;value={
A:a,
B:b1,
C:c
}
void Consumer::drain()
执行m_toSync中的任务。
void Consumer::drain()
{
if (!m_toSync.empty())
m_orch->doTask(*this);
}
class Orch
class Orch
{
public:
//每个orch都会连接到数据库,以及其需要订阅的表名,和订阅该表产生的事件的优先级
//以默认优先级订阅一个table
Orch(DBConnector *db, const string tableName, int pri = default_orch_pri);
//以默认优先级订阅多个table
Orch(DBConnector *db, const vector<string> &tableNames);
//订阅多个table,指明每个table的优先级
Orch(DBConnector *db, const vector<table_name_with_pri_t> &tableNameWithPri);
//连接多个数据库
Orch(const vector<TableConnector>& tables);
virtual ~Orch();
// 获取该orch的所有epoll事件
vector<Selectable*> getSelectables();
/* Iterate all consumers in m_consumerMap and run doTask(Consumer) */
// 执行该orch中所有的consumers中的m_sync中的任务
void doTask();
/* Run doTask against a specific executor */
// 任务的来源可以是consumer,NotificationConsumer,SelectableTimer
virtual void doTask(Consumer &consumer) = 0;
virtual void doTask(NotificationConsumer &consumer) { }
virtual void doTask(SelectableTimer &timer) { }
/* TODO: refactor recording */
static void recordTuple(Consumer &consumer, KeyOpFieldsValuesTuple &tuple);
protected:
// 消费者map,一个orch可以订阅多个table,key为tableName,value为Executor
ConsumerMap m_consumerMap;
// 与调试相关
static void logfileReopen();
string dumpTuple(Consumer &consumer, KeyOpFieldsValuesTuple &tuple);
ref_resolve_status resolveFieldRefValue(type_map&, const string&, KeyOpFieldsValuesTuple&, sai_object_id_t&);
bool parseIndexRange(const string &input, sai_uint32_t &range_low, sai_uint32_t &range_high);
bool parseReference(type_map &type_maps, string &ref, string &table_name, string &object_name);
ref_resolve_status resolveFieldRefArray(type_map&, const string&, KeyOpFieldsValuesTuple&, vector<sai_object_id_t>&);
/* Note: consumer will be owned by this class */
// 内部函数添加一个Executor,给addConsumer使用
void addExecutor(string executorName, Executor* executor);
Executor *getExecutor(string executorName);
private:
// 添加一个消费者
void addConsumer(DBConnector *db, string tableName, int pri = default_orch_pri);
};
void Orch::addConsumer(……)
void Orch::addExecutor(string executorName, Executor* executor)
{
m_consumerMap.emplace(std::piecewise_construct,
std::forward_as_tuple(executorName),
std::forward_as_tuple(executor));
}
//添加一个消费者
void Orch::addConsumer(DBConnector *db, string tableName, int pri)
{
if (db->getDbId() == CONFIG_DB || db->getDbId() == STATE_DB)
{
addExecutor(tableName, new Consumer(new SubscriberStateTable(db, tableName, TableConsumable::DEFAULT_POP_BATCH_SIZE, pri), this));
}
else
{
addExecutor(tableName, new Consumer(new ConsumerStateTable(db, tableName, gBatchSize, pri), this));
}
}
void Orch::doTask(……)
执行本orch中的每一个消费者m_toSync中的task,不管该task是否本次从redis中读取还是以前未处理完毕的。
void Orch::doTask()
{
for(auto &it : m_consumerMap)
{
it.second->drain();
}
}
class Orch2
orch2是在orch的基础上的一个封装,代码的可读性增强。
class Orch2 : public Orch
{
public:
Orch2(DBConnector *db, const std::string& tableName, Request& request, int pri=default_orch_pri)
: Orch(db, tableName, pri), request_(request)
{
}
protected:
virtual void doTask(Consumer& consumer);
virtual bool addOperation(const Request& request)=0;
virtual bool delOperation(const Request& request)=0;
private:
Request& request_;
};
void Orch2::doTask
void Orch2::doTask(Consumer &consumer)
{
SWSS_LOG_ENTER();
auto it = consumer.m_toSync.begin();
while (it != consumer.m_toSync.end())
{
bool erase_from_queue = true;
try
{
request_.parse(it->second);
auto op = request_.getOperation();
if (op == SET_COMMAND)
{
erase_from_queue = addOperation(request_);
}
else if (op == DEL_COMMAND)
{
erase_from_queue = delOperation(request_);
}
else
{
SWSS_LOG_ERROR("Wrong operation. Check RequestParser: %s", op.c_str());
}
}
catch (const std::invalid_argument& e)
{
SWSS_LOG_ERROR("Parse error: %s", e.what());
}
catch (const std::logic_error& e)
{
SWSS_LOG_ERROR("Logic error: %s", e.what());
}
catch (const std::exception& e)
{
SWSS_LOG_ERROR("Exception was catched in the request parser: %s", e.what());
}
catch (...)
{
SWSS_LOG_ERROR("Unknown exception was catched in the request parser");
}
request_.clear();
//执行成功,那么从m_tosync中删除,否则执行下一个task
if (erase_from_queue)
{
it = consumer.m_toSync.erase(it);
}
else
{
++it;
}
}
}