IO多路复用

太空铝阳台洗衣柜 · 发表于 2023-2-26 21:58:04

1.IO多路复用的概念

单线程或单进程同时监测若干个文件描述符是否可以执行IO操作的能力。

2.为什么出现IO多路复用

服务器需要维护N多个与客户端之间的socketfd；并且在receive之前需要知道数据知否出现---》组件IO多路复用技术出现---》解决检测服务器端N多个fd的状态
- Tcp是有连接的，Udp是无连接---》上述情况出现在Tcp连接情况
IO多路复用的三种方案：select/poll/epoll
- select(fds+1, rds, wds, timeout)
- poll(fds, nfd, timeout)
- epoll
  - epoll_create(size/flags)--》创建根节点---》epoll实例
  - epoll_ctl(fd, op, events)---》挂载fd节点
  - epoll_wait(fd, events, maxevents, timeout)---》检测
Tcp建立连接---》需要的fds放入IO多路复用中管理---》检测fd缓存状态---》receive
- receive(fd,buff,sizeof(buff),0)

3.IO多路复用区别特点

selectpollepoll性能随IO处理数增加，性能逐渐下降；并且连接数有限制随IO处理数增加，性能逐渐下降随IO处理数增加，性能基本上不会下降连接数最大连接数1024；处理1024以上则需要修改宏FD_SETSIZE，重新编译无限制无限制数据结构不确定，有数组的可能，但是为线性结构不确定，有链表的可能，但是为线性结构红黑树，队列处理方式线性轮询线性轮询回调callback内存拷贝所有fds从用户空间和内核空间来回拷贝**所有fds从用户空间和内核空间来回拷贝epoll_wait()只需要从就绪队列上拷贝由内核空间到用户空间使用复杂度低低中时间复杂度O(n)O(n)O()

select/poll
- 拷贝fds--》从用户空间至内核
- for(;;)进行遍历判断
  - select：进行分类rds，wds，eds
  - poll：pollfd维护自身状态
- 从内核拷贝至用户空间
epoll
- epoll_create--->创建一个根节点
- epoll_ctl---》挂载节点
- epoll_wait---》从内核到用户只拷贝满足条件的节点，从就绪队列中拷贝
在哪些场景下 select比epoll更合适？
- IO数量不多，且使用多线程多进程的情况使用select比epoll更合适
  - epoll使用红黑树，必须需要加锁，消耗的资源更多

红黑树：
等待补充链接
3.IO多路复用源码

select：kern_select---》core_sys_select---》do_select---linux-6.0.2\fs\select.c
- 1. static int kern_select(int n, fd_set __user *inp, fd_set __user *outp,fd_set __user *exp, struct __kernel_old_timeval __user *tvp){
  2. struct timespec64 end_time, *to = NULL;
  3. struct __kernel_old_timeval tv;
  4. int ret;
  5. if (tvp) {
  6. //copy_from_user从用户空间拷贝fds
  7. if (copy_from_user(&tv, tvp, sizeof(tv)))
  8. return -EFAULT;
  9. to = &end_time;
  10. if (poll_select_set_timeout(to,
  11. tv.tv_sec + (tv.tv_usec / USEC_PER_SEC),
  12. (tv.tv_usec % USEC_PER_SEC) * NSEC_PER_USEC))
  13. return -EINVAL;
  14. }
  15. //调用select--》core_sys_select
  16. ret = core_sys_select(n, inp, outp, exp, to);
  17. return poll_select_finish(&end_time, tvp, PT_TIMEVAL, ret);
  18. }
  复制代码
- 1. int core_sys_select(int n, fd_set __user *inp, fd_set __user *outp,
  2. fd_set __user *exp, struct timespec64 *end_time)
  3. {
  4. ......
  5. //加读锁,读取文件数量
  6. rcu_read_lock();
  7. fdt = files_fdtable(current->files);
  8. max_fds = fdt->max_fds;
  9. rcu_read_unlock();
  10. ......
  11. //select底层处理---》do_select
  12. ret = do_select(n, &fds, end_time);
  14. if (ret < 0)
  15. goto out;
  16. if (!ret) {
  17. ret = -ERESTARTNOHAND;
  18. if (signal_pending(current))
  19. goto out;
  20. ret = 0;
  21. }
  22. ......
  23. out_nofds:
  24. return ret;
  25. }
  复制代码
- 1. static int do_select(int n, fd_set_bits *fds, struct timespec64 *end_time)
  2. {
  3. .....
  4. //读锁，读取文件数量
  5. rcu_read_lock();
  6. retval = max_select_fd(n, fds);
  7. rcu_read_unlock();
  8. .....
  9. //通过循环遍历，主要流程流程
  10. for (;;) {
  11. unsigned long *rinp, *routp, *rexp, *inp, *outp, *exp;
  12. bool can_busy_loop = false;
  14. inp = fds->in; outp = fds->out; exp = fds->ex;
  15. rinp = fds->res_in; routp = fds->res_out; rexp = fds->res_ex;
  17. for (i = 0; i < n; ++rinp, ++routp, ++rexp) {
  18. unsigned long in, out, ex, all_bits, bit = 1, j;
  19. unsigned long res_in = 0, res_out = 0, res_ex = 0;
  20. __poll_t mask;
  22. in = *inp++; out = *outp++; ex = *exp++;
  23. all_bits = in | out | ex;
  24. if (all_bits == 0) {
  25. i += BITS_PER_LONG;
  26. continue;
  27. }
  29. for (j = 0; j < BITS_PER_LONG; ++j, ++i, bit <<= 1) {
  30. struct fd f;
  31. if (i >= n)
  32. break;
  33. if (!(bit & all_bits))
  34. continue;
  35. mask = EPOLLNVAL;
  36. f = fdget(i);
  37. if (f.file) {
  38. wait_key_set(wait, in, out, bit,
  39. busy_flag);
  40. mask = vfs_poll(f.file, wait);
  42. fdput(f);
  43. }
  44. if ((mask & POLLIN_SET) && (in & bit)) {
  45. res_in |= bit;
  46. retval++;
  47. wait->_qproc = NULL;
  48. }
  49. if ((mask & POLLOUT_SET) && (out & bit)) {
  50. res_out |= bit;
  51. retval++;
  52. wait->_qproc = NULL;
  53. }
  54. if ((mask & POLLEX_SET) && (ex & bit)) {
  55. res_ex |= bit;
  56. retval++;
  57. wait->_qproc = NULL;
  58. }
  59. if (retval) {
  60. can_busy_loop = false;
  61. busy_flag = 0;
  62. } else if (busy_flag & mask)
  63. can_busy_loop = true;
  64. }
  65. if (res_in)
  66. *rinp = res_in;
  67. if (res_out)
  68. *routp = res_out;
  69. if (res_ex)
  70. *rexp = res_ex;
  71. cond_resched();
  72. }
  73. wait->_qproc = NULL;
  74. if (retval || timed_out || signal_pending(current))
  75. break;
  76. if (table.error) {
  77. retval = table.error;
  78. break;
  79. }
  80. if (can_busy_loop && !need_resched()) {
  81. if (!busy_start) {
  82. busy_start = busy_loop_current_time();
  83. continue;
  84. }
  85. if (!busy_loop_timeout(busy_start))
  86. continue;
  87. }
  88. busy_flag = 0;
  89. if (end_time && !to) {
  90. expire = timespec64_to_ktime(*end_time);
  91. to = &expire;
  92. }
  94. if (!poll_schedule_timeout(&table, TASK_INTERRUPTIBLE,
  95. to, slack))
  96. timed_out = 1;
  97. }
  99. poll_freewait(&table);
  101. return retval;
  102. }
  复制代码
poll：do_sys_poll---》do_poll---linux-6.0.2\fs\select.c
- 1. static int do_sys_poll(struct pollfd __user *ufds, unsigned int nfds,struct timespec64 *end_time)
  2. {
  3. ......
  4. len = min_t(unsigned int, nfds, N_STACK_PPS);
  5. for (;;) {
  6. .....
  7. //copy_from_user从用户空间拷贝fds
  8. if (copy_from_user(walk->entries, ufds + nfds-todo,
  9. sizeof(struct pollfd) * walk->len))
  10. goto out_fds;
  11. .....
  12. }
  13. .....
  14. out_fds:
  15. //没有拷贝成功
  16. .....
  17. return err;
  18. .....
  19. }
  复制代码
- 1. static int do_poll(struct poll_list *list, struct poll_wqueues *wait,struct timespec64 *end_time)
  2. {
  3. ......
  4. //poll底层处理流程
  5. for (;;) {
  6. struct poll_list *walk;
  7. bool can_busy_loop = false;
  8. //这个写法，似乎是链表
  9. for (walk = list; walk != NULL; walk = walk->next) {
  10. struct pollfd * pfd, * pfd_end;
  12. pfd = walk->entries;
  13. pfd_end = pfd + walk->len;
  14. for (; pfd != pfd_end; pfd++) {
  15. if (do_pollfd(pfd, pt, &can_busy_loop,
  16. busy_flag)) {
  17. count++;
  18. pt->_qproc = NULL;
  19. /* found something, stop busy polling */
  20. busy_flag = 0;
  21. can_busy_loop = false;
  22. }
  23. }
  24. }
  25. pt->_qproc = NULL;
  26. if (!count) {
  27. count = wait->error;
  28. if (signal_pending(current))
  29. count = -ERESTARTNOHAND;
  30. }
  31. if (count || timed_out)
  32. break;
  33. if (can_busy_loop && !need_resched()) {
  34. if (!busy_start) {
  35. busy_start = busy_loop_current_time();
  36. continue;
  37. }
  38. if (!busy_loop_timeout(busy_start))
  39. continue;
  40. }
  41. busy_flag = 0;
  43. if (end_time && !to) {
  44. expire = timespec64_to_ktime(*end_time);
  45. to = &expire;
  46. }
  48. if (!poll_schedule_timeout(wait, TASK_INTERRUPTIBLE, to, slack))
  49. timed_out = 1;
  50. }
  51. return count;
  52. }
  复制代码
epoll：do_epoll_wait---》ep_poll--linux-6.0.2\fs\eventpoll.c
- 1. static int do_epoll_wait(int epfd, struct epoll_event __user *events,int maxevents, struct timespec64 *to)
  2. {
  3. .....
  4. //调用ep_poll
  5. error = ep_poll(ep, events, maxevents, to);
  7. error_fput:
  8. fdput(f);
  9. return error;
  10. }
  复制代码
- 1. static int ep_poll(struct eventpoll *ep, struct epoll_event __user *events,int maxevents, struct timespec64 *timeout)
  2. {
  3. int res, eavail, timed_out = 0;
  4. u64 slack = 0;
  5. wait_queue_entry_t wait;
  6. ktime_t expires, *to = NULL;
  8. lockdep_assert_irqs_enabled();
  10. if (timeout && (timeout->tv_sec | timeout->tv_nsec)) {
  11. slack = select_estimate_accuracy(timeout);
  12. to = &expires;
  13. *to = timespec64_to_ktime(*timeout);
  14. } else if (timeout) {
  15. timed_out = 1;
  16. }
  17. eavail = ep_events_available(ep);
  19. while (1) {
  20. if (eavail) {
  21. res = ep_send_events(ep, events, maxevents);
  22. if (res)
  23. return res;
  24. }
  26. if (timed_out)
  27. return 0;
  29. eavail = ep_busy_loop(ep, timed_out);
  30. if (eavail)
  31. continue;
  33. if (signal_pending(current))
  34. return -EINTR;
  35. init_wait(&wait);
  36. wait.func = ep_autoremove_wake_function;
  38. write_lock_irq(&ep->lock);
  39. __set_current_state(TASK_INTERRUPTIBLE);
  40. eavail = ep_events_available(ep);
  41. if (!eavail)
  42. __add_wait_queue_exclusive(&ep->wq, &wait);
  44. write_unlock_irq(&ep->lock);
  46. if (!eavail)
  47. timed_out = !schedule_hrtimeout_range(to, slack,
  48. HRTIMER_MODE_ABS);
  49. __set_current_state(TASK_RUNNING);
  50. eavail = 1;
  52. if (!list_empty_careful(&wait.entry)) {
  53. write_lock_irq(&ep->lock);
  54. if (timed_out)
  55. eavail = list_empty(&wait.entry);
  56. __remove_wait_queue(&ep->wq, &wait);
  57. write_unlock_irq(&ep->lock);
  58. }
  59. }
  60. }
  复制代码

4.参考资料

一文搞懂select、poll和epoll区别 - 知乎 (zhihu.com)
深入浅出理解select、poll、epoll的实现 - 知乎 (zhihu.com)
Linux内核源代码查看及分析方法 - 知乎 (zhihu.com)

来源:https://www.cnblogs.com/zqurgy/p/17156664.html
免责声明：由于采集信息均来自互联网，如果侵犯了您的权益，请联系我们【E-Mail:cb@itdo.tech】我们会及时删除侵权内容，谢谢合作！

IO多路复用

本帖子中包含更多资源