sonic orch调度系统(1)----select

常见的服务器模型有多进程模型，多线程，IO多路复用，协程等模型。sonic的核心守护进程orchagent采用的是IO多路复用模型，早期的sonic采用的是select实现多路复用，后面的版本采用的是epoll。使用select(跟多路复用的select名字一样)类对底层进行了封装，屏蔽了差异。

class Selectable

事件基类，描述了epoll事件，可以是读，写，异常等事件。该结构对通用epoll事件进行了封装，真实事件通过该类派生出来，比如redis数据库事件：class RedisSelect : public Selectable；netlink事件：class NetLink : public Selectable；通知：class NotificationConsumer : public Selectable，orch执行单元：class Executor : public Selectable，定时器：class SelectableTimer : public Selectable等。

class Selectable
{
public:Selectable(int pri = 0) : m_priority(pri),m_last_used_time(std::chrono::steady_clock::now()) {lastusedsequence = g_lastusedsequence++;}virtual ~Selectable() = default;/* return file handler for the Selectable */virtual int getFd() = 0;/* Read all data from the fd assicaited with Selectable */virtual void readData() = 0;/* true if Selectable has data in its cache */// 判断是否还有数据，如果有放入就绪事件setvirtual bool hasCachedData(){return false;}/* true if Selectable was initialized with data */// 判断是否需要读取初始数据virtual bool initializedWithData(){return false;}/* run this function after every read */// 更新事件数virtual void updateAfterRead(){}int getPri() const{return m_priority;}private:friend class Select;//友元类为Select// only Select class can access and update m_last_used_timestd::chrono::time_point<std::chrono::steady_clock> getLastUsedTime() const{return m_last_used_time;}// 最后使用序列号unsigned long getLastUsedsequence() const{return lastusedsequence;}// 跟新最后使用序列号void updateLastUsedTime(){m_last_used_time = std::chrono::steady_clock::now();lastusedsequence = g_lastusedsequence++;}// 优先级，实现基于优先级调度int m_priority; // defines priority of Selectable inside Select// higher value is higher prioritystd::chrono::time_point<std::chrono::steady_clock> m_last_used_time;unsigned long lastusedsequence;//上次使用序列号static unsigned long g_lastusedsequence;//全局基准序列号，用于对同优先级业务进行公平调度
};

class Select

class Select
{
public:Select();~Select();/* Add object for select 给epoll添加一个事件 */void addSelectable(Selectable *selectable);/* Remove object from select 删除一个epoll事件 */void removeSelectable(Selectable *selectable);/* Add multiple objects for select 添加多个epoll事件 */void addSelectables(std::vector<Selectable *> selectables);enum {//返回的事件类型OBJECT = 0,ERROR = 1,TIMEOUT = 2,};//执行epoll int select(Selectable **c, unsigned int timeout = std::numeric_limits<unsigned int>::max());int select(std::vector<Selectable *> &vc, unsigned int timeout = std::numeric_limits<unsigned int>::max());private://epoll事件比较函数，通过该函数实现事件的优先级struct cmp{bool operator()(const Selectable* a, const Selectable* b) const{/* Choose Selectable with highest priority first */if (a->getPri() > b->getPri())return true;else if (a->getPri() < b->getPri())return false;/* if the priorities are equal *//* use Selectable which was selected later */if (a->getLastUsedsequence() < b->getLastUsedsequence())return true;else if (a->getLastUsedsequence() > b->getLastUsedsequence())return false;/* when a == b */return false;}};//epoll轮询函数int poll_descriptors(Selectable **c, unsigned int timeout);int poll_descriptors(std::vector<Selectable *> &vc, unsigned int timeout);int m_epoll_fd;//epoll句柄std::unordered_map<int, Selectable *> m_objects;//监听的事件句柄与其对应的selectable之间的关系std::set<Selectable *, Select::cmp> m_ready;//已经就绪的事件集合，提供了比较函数，从而实现优先级调度
};

Select::Select()

Select::Select()
{m_epoll_fd = ::epoll_create1(0);//创建epoll句柄if (m_epoll_fd == -1){std::string error = std::string("Select::constructor:epoll_create1: error=("+ std::to_string(errno) + "}:"+ strerror(errno));throw std::runtime_error(error);}
}

Select::~Select()

Select::~Select()
{(void)::close(m_epoll_fd);
}

void Select::addSelectable(Selectable *selectable)

void Select::addSelectable(Selectable *selectable)
{const int fd = selectable->getFd();if(m_objects.find(fd) != m_objects.end())//已经添加了该事件，退出{SWSS_LOG_WARN("Selectable is already added to the list, ignoring.");return;}m_objects[fd] = selectable;if (selectable->initializedWithData())//是否已经有数据可读，读出已有的数据{m_ready.insert(selectable);}//添加可读事件struct epoll_event ev = {.events = EPOLLIN,.data = { .fd = fd, },};int res = ::epoll_ctl(m_epoll_fd, EPOLL_CTL_ADD, fd, &ev);if (res == -1){std::string error = std::string("Select::add_fd:epoll_ctl: error=("+ std::to_string(errno) + "}:"+ strerror(errno));throw std::runtime_error(error);}
}

void Select::removeSelectable(Selectable *selectable)

void Select::removeSelectable(Selectable *selectable)
{const int fd = selectable->getFd();m_objects.erase(fd);m_ready.erase(selectable);//从epoll中删除事件int res = ::epoll_ctl(m_epoll_fd, EPOLL_CTL_DEL, fd, NULL);if (res == -1){std::string error = std::string("Select::del_fd:epoll_ctl: error=("+ std::to_string(errno) + "}:"+ strerror(errno));throw std::runtime_error(error);}
}

void Select::addSelectables(vector<Selectable *> selectables)

void Select::addSelectables(vector<Selectable *> selectables)
{for(auto it : selectables)//添加多个事件{addSelectable(it);}
}

int Select::poll_descriptors(......)

提取一个就绪事件

int Select::poll_descriptors(Selectable **c, unsigned int timeout)
{int sz_selectables = static_cast<int>(m_objects.size());std::vector<struct epoll_event> events(sz_selectables);int ret;//阻塞等待事件发生，发生错误或者被中断打断则继续监听，发生事件则执行事件do{ret = ::epoll_wait(m_epoll_fd, events.data(), sz_selectables, timeout);}while(ret == -1 && errno == EINTR); // Retry the select if the process was interrupted by a signalif (ret < 0)return Select::ERROR;//遍历每一个发生的事件for (int i = 0; i < ret; ++i){int fd = events[i].data.fd;Selectable* sel = m_objects[fd];//获取事件描述符sel->readData();//读取数据m_ready.insert(sel);//插入就绪集合}//存在就绪事件if (!m_ready.empty()){auto sel = *m_ready.begin();*c = sel;m_ready.erase(sel);// we must update clock only when the selector out of the m_ready// otherwise we break invariant of the m_ready// 更新该事件的使用时间，使用事件会作为事件优先级进行使用，越频繁的优先级越低，从而避免同优先级的事件// 饿死sel->updateLastUsedTime();// 有数据，依然放入已经就绪集合if (sel->hasCachedData()){// reinsert Selectable back to the m_ready set, when there're more messages in the cachem_ready.insert(sel);}// 对数据进行刷新，如果该句柄只发生了一次事件，那么这里会进行减1，下次m_ready中将不会存在该sel。// 仔细分析了sonic的selectable的实现，这里是有bug的，会造成大量的空转。sel->updateAfterRead();return Select::OBJECT;}return Select::TIMEOUT;
}

int Select::poll_descriptors(std::vector<Selectable *> &vc, unsigned int timeout)

提取多个就绪事件，该函数是在上面的函数的基础上的改进。只提取一个事件将会造成"饿死和胀死"的问题。由于m_ready是有序队列，对于高优先的事件总是会被优先提取，如果高优先级的事件依赖于低优先级事件的话，会造成高优先级的业务一直被调度，但是缺少依赖条件而不能执行业务，低优先级业务总是得不到调度，形成死锁问题。同时提取所有就绪事件可以解决高低优先级死锁问题。

int Select::poll_descriptors(std::vector<Selectable *> &vc, unsigned int timeout)
{int sz_selectables = static_cast<int>(m_objects.size());std::vector<struct epoll_event> events(sz_selectables);int ret;do{ret = ::epoll_wait(m_epoll_fd, events.data(), sz_selectables, timeout);}while(ret == -1 && errno == EINTR); // Retry the select if the process was interrupted by a signalif (ret < 0)return Select::ERROR;for (int i = 0; i < ret; ++i){int fd = events[i].data.fd;Selectable* sel = m_objects[fd];sel->readData();m_ready.insert(sel);}auto iter = m_ready.begin();while(iter !=m_ready.end()){auto sel = *iter;vc.push_back(sel);iter = m_ready.erase(iter);sel->updateLastUsedTime();}for(auto se:vc){if (se->hasCachedData()){m_ready.insert(se);}se->updateAfterRead();}if(!vc.empty()){return Select::OBJECT;}return Select::TIMEOUT;
}

int Select::select(Selectable **c, unsigned int timeout)

int Select::select(Selectable **c, unsigned int timeout)
{SWSS_LOG_ENTER();int ret;*c = NULL;if (timeout == numeric_limits<unsigned int>::max())timeout = -1;/* check if we have some data */ret = poll_descriptors(c, 0);/* return if we have data, we have an error or desired timeout was 0 */if (ret != Select::TIMEOUT || timeout == 0)return ret;/* wait for data */ret = poll_descriptors(c, timeout);return ret;
}

int Select::select(std::vector<Selectable *> &vc, unsigned int timeout)

int Select::select(std::vector<Selectable *> &vc, unsigned int timeout)
{SWSS_LOG_ENTER();int ret;if (timeout == numeric_limits<unsigned int>::max())timeout = -1;/* check if we have some data */ret = poll_descriptors(vc, 0);/* return if we have data, we have an error or desired timeout was 0 */if (ret != Select::TIMEOUT || timeout == 0)return ret;/* wait for data */ret = poll_descriptors(vc, timeout);return ret;}