sonic orch调度系统之----orchagent

​ sonic核心守护线程orchagent以orch为单位进行资源管理，一个orch包含了一组类似的资源；orchagent调度系统以Executor为调度单位，调度实体有Consumer，ExecutableTimer等，本文分析一下sonic调度细节。

class OrchDaemon

orchagent以OrchDaemon做为核心类进行描述。

class OrchDaemon
{
public:
    OrchDaemon(DBConnector *, DBConnector *, DBConnector *);
    ~OrchDaemon();

    bool init();//初始化进程
    void start();//启动调度系统
private:
    //链接了三个数据库
    DBConnector *m_applDb;
    DBConnector *m_configDb;
    DBConnector *m_stateDb;
    //包含全部的orch
    std::vector<Orch *> m_orchList;
    //建立的select多路异步IO控制块
    Select *m_select;
    //将asic_db的pipe进行flush，不在等待。
    void flush();
};

bool OrchDaemon::init()

初始化orchagent执行环境。

bool OrchDaemon::init()
{
    SWSS_LOG_ENTER();

    ......
        
    //链接数据库
    TableConnector confDbAclTable(m_configDb, CFG_ACL_TABLE_NAME);
    TableConnector confDbAclRuleTable(m_configDb, CFG_ACL_RULE_TABLE_NAME);
    TableConnector stateDbLagTable(m_stateDb, STATE_LAG_TABLE_NAME);

    vector<TableConnector> acl_table_connectors = {
        confDbAclTable,
        confDbAclRuleTable,
        stateDbLagTable
    };

    ......
    
    //收集对应的orch
    m_orchList = { switch_orch, gCrmOrch, gBufferOrch, gPortsOrch, intfs_orch, gNeighOrch, gRouteOrch, copp_orch, tunnel_decap_orch, qos_orch, mirror_orch, gAclOrch, gFdbOrch, vrf_orch };
    //建立多路事件控制块
    m_select = new Select();

    ......
    //除了上面的orch外，其它代码还添加了一些orch，这里再也不列出

    return true;
}

void OrchDaemon::flush()

将asic_db的生产者的pipeline清空

/* Flush redis through sairedis interface */
void OrchDaemon::flush()
{
    SWSS_LOG_ENTER();

    sai_attribute_t attr;
    attr.id = SAI_REDIS_SWITCH_ATTR_FLUSH;
    sai_status_t status = sai_switch_api->set_switch_attribute(gSwitchId, &attr);
    if (status != SAI_STATUS_SUCCESS)
    {
        SWSS_LOG_ERROR("Failed to flush redis pipeline %d", status);
        exit(EXIT_FAILURE);
    }
}

void OrchDaemon::start()

void OrchDaemon::start()
{
    SWSS_LOG_ENTER();
    //遍历每个orch，将每个orch中的全部关心的事件加入epoll中，基本一个Executor为一个事件
    for (Orch *o : m_orchList)
    {
        m_select->addSelectables(o->getSelectables());
    }
    //进入dead loop
    while (true)
    {
        Selectable *s;
        int ret;
        // 进行epoll阻塞监听
        ret = m_select->select(&s, SELECT_TIMEOUT);
        // 错误事件，继续监听
        if (ret == Select::ERROR)
        {
            SWSS_LOG_NOTICE("Error: %s!\n", strerror(errno));
            continue;
        }
        //超时事件
        if (ret == Select::TIMEOUT)
        {
            /* Let sairedis to flush all SAI function call to ASIC DB.
             * Normally the redis pipeline will flush when enough request
             * accumulated. Still it is possible that small amount of
             * requests live in it. When the daemon has nothing to do, it
             * is a good chance to flush the pipeline  
             * 确保redis-pipeline中的少许请求在10秒后可以获得处理，不在积累请求
             */
            flush();
            continue;
        }
        //获取触发epoll事件的Executor
        auto *c = (Executor *)s;
        //对于Consumer来讲，执行数据库操做，将redis中的通知转换到m_tosync中，而且执行如下m_tosync中的task
        c->execute();

        /* After each iteration, periodically check all m_toSync map to
         * execute all the remaining tasks that need to be retried. */

        /* TODO: Abstract Orch class to have a specific todo list */
        // 执行其它的orch中遗留的任务
        for (Orch *o : m_orchList)
            o->doTask();

    }
}

实例分析

下面咱们以orchagent处理的最多的事件:ConsumerStateTable来分析一下orchagent调度系统。

ConsumerStateTable即orchagent订阅的app_db事件。格式以下所示：

"SADD" "INTF_TABLE_KEY_SET" "PortChannel1:1.1.1.1/8"        #在集合INTF_TABLE_KEY_SET中增长一个key
"HSET" "INTF_TABLE:PortChannel1:1.1.1.1/8" "scope" "global" 
"HSET" "INTF_TABLE:PortChannel1:1.1.1.1/8" "family" "IPv4"
"PUBLISH" "INTF_TABLE_CHANNEL" "G"

• 当生产者发送命令"PUBLISH" "INTF_TABLE_CHANNEL" "G",orchagent从epoll中被唤醒，而后从对应的redis客户端句柄中调用RedisSelect::readData()进行数据读取该命令的应答，应答以下所示：

1) "message"
2) "INTF_TABLE_CHANNEL"
3) "G"

• 每收到一个应答都会对RedisSelect::m_queueLength进行加1，该应答只有一个信息"G",只起到通知做用。因此若是生产者同时写入了大量的app_db事件，某一个epoll唤醒，RedisSelect::readData()能够读出大量的应答，从而致使m_queueLength大于1。
• 读完数据后将该事件Selectable添加到Select：m_ready中。同时经过调用RedisSelect::updateAfterRead()对RedisSelect::m_queueLength进行减1，这里只减去了1，而没有根据实际处理的状况减去对应的值，这里存在增长多个值减去一个值的状况。只要m_queueLength不为1，就不会将Selectable从Select：m_ready删除。
• Select::select函数会返回Select：m_ready中全部的Selectable。
• 遍历每个selectable，在函数Consumer::execute()中调用TableEntryPoppable::pops使用脚本consumer_state_table_pops.lua处理INTF_TABLE_KEY_SET中的key，每次最多处理128个key。

local ret = {}
local keys = redis.call('SPOP', KEYS[1], ARGV[1])--一次处理128个key
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})
end
return ret

脚本返回的内容将会是：

INTF_TABLE:PortChannel1:1.1.1.1/8:{
    "scope":"global",
    "family": "IPv4"
}

pops而后对上面的内容进行加工成以下格式：

std::tuple<std::string, std::string, std::vector<FieldValueTuple> >

即最后返回：

"SET", "INTF_TABLE:PortChannel1:1.1.1.1/8", <"scope":"global","family": "IPv4">

• Consumer::execute()会对前面返回的值添加到Consumer::m_toSync,这里会进行合并。
• 最后OrchDaemon::start()函数会执行每个orch的Consumer::m_toSync中的task。

sonic调度系统的缺陷

sonic调度系统过于简单，没法处理大规模，逻辑复杂的业务，效率很是低下。

1. 当超规格下发配置的时候，会致使m_toSync中由于资源不足而驻留大量的task。OrchDaemon::start()每一次事件触发都会遍历全部orch的m_toSync中的task，形成无效task频繁执行，引起orchagent发生震荡。必须给Consumer增长一些标记，代表当前Consumer处于资源不足状态，暂不执行m_toSync中的task
2. 当大规模下发配置的时候，若是本配置的依赖尚未下发，会致使m_toSync中由于依赖不知足而驻留大量的task。引起orchagent震荡问题。
3. 当大规模下删除配置的时候，若是本配置引用尚未删除，会致使m_toSync中由于被引用不能删除而驻留大量的task。引起orchagent震荡问题。
4. orchagent在处理task的时候没有考虑事件处理顺序，应该首先执行DEL操做，而后再执行SET操做。对于DEL操做，应该先删除低优先级的Consumer，对于SET操做来讲，应该先处理高优先级的Consumer。二者顺序相反。