• 深入理解Linux内核select多路复用原理

    写在前面：本文以Linux2.6.0的内核源码进行讲解，使用x86 32位机讲解。

    多路复用原理

    讲多路复用的原理，那么一定先要讲没有多路复用的弊端。传统的阻塞式，进程一旦io读写就开始阻塞，效率太低，导致整个系统的吞吐量急剧下降。那么，就有人要思考，居然要阻塞，那么我就使用线程去阻塞，那么一个系统的io操作如此之多，要多少个线程才能满足呢？线程是不是会占用资源？当线程多起来后分时复用的CPU需要一直切换上下文，切换上下文是不是在浪费CPU资源呢，会导致大量的时间在切换上下文？

    所以就引入了多路复用，把需要io读写的操作叫做一个事件（多个事件，所以叫多路），把这些io事件维护在一个队列中，并且每个事件都绑定上一个回调函数（回调函数负责唤醒当前阻塞的进程），而这些事件的绑定回调函数和维护队列的操作都是交给一个进程来做（所以叫复用）。当底层网卡没有触发读写事件的时候，这个进程就去阻塞等待（让出CPU的使用权），当底层网卡建立好连接后会发出中断信号，此时中断处理函数就会回调事件对应的回调函数，而回调函数就是让进程唤醒，唤醒后，进程会去遍历所有事件，得到那个准备好了的事件，最终把所有事件返回给用户态（所以用户态还需要遍历才能得到准备好的事件）。

    考虑到，文字描述比较难懂，特意画了一张流程图帮助读者来理解（尽力了...）

    此多路复用只是内核中select的流程，poll和epoll的流程略有区别，poll和epoll后续会有文章来细讲。

     

     

    select系统调用源码讲解：

    如何从用户态进入内核态，执行系统调用的流程，这里就不进行讲解，那么直接讲解sys_select的方法。       

    其次，在Linux内核中，VFS万物皆文件的思想，一言以蔽之：不管是普通文件，网络、外设设备等等都是抽象成一个文件，而文件在内核中的实现file结构体返回给用户态不安全，所以都是返回的fd下标给用户态。

    sys_select源码:

    1. // n是事件数量
    2. // inp是输入事件的位图
    3. // outp是输出事件的位图
    4. // exp是不感兴趣事件的位图
    5. // tvp是计时器
    6. asmlinkage long
    7. sys_select(int n, fd_set __user *inp, fd_set __user *outp, fd_set __user *exp, struct timeval __user *tvp)
    8. {
    9. 	fd_set_bits fds;
    10. 	char *bits;
    11. 	long timeout;
    12. 	int ret, size, max_fdset;
    13.  
    14. 	timeout = MAX_SCHEDULE_TIMEOUT;
    15. 	if (tvp) {
    16. 		time_t sec, usec;
    17.  
    18. 		if ((ret = verify_area(VERIFY_READ, tvp, sizeof(*tvp)))
    19. 		    || (ret = __get_user(sec, &tvp->tv_sec))
    20. 		    || (ret = __get_user(usec, &tvp->tv_usec)))
    21. 			goto out_nofds;
    22.  
    23. 		ret = -EINVAL;
    24. 		if (sec < 0 || usec < 0)
    25. 			goto out_nofds;
    26.  
    27. 		if ((unsigned long) sec < MAX_SELECT_SECONDS) {
    28. 			timeout = ROUND_UP(usec, 1000000/HZ);
    29. 			timeout += sec * (unsigned long) HZ;
    30. 		}
    31. 	}
    32.  
    33. 	ret = -EINVAL;
    34. 	if (n < 0)
    35. 		goto out_nofds;
    36.  
    37. 	/* max_fdset can increase, so grab it once to avoid race */
    38. 	max_fdset = current->·files->max_fdset;
    39. 	if (n > max_fdset)
    40. 		n = max_fdset;
    41.  
    42. 	/*
    43. 	 * We need 6 bitmaps (in/out/ex for both incoming and outgoing),
    44. 	 * since we used fdset we need to allocate memory in units of
    45. 	 * long-words. 
    46. 	 */
    47. 	ret = -ENOMEM;
    48.  
    49. 	// 根据n的数量，算出需要多少字节（也就是多少位）
    50. 	size = FDS_BYTES(n);
    51.  
    52. 	// 开辟4 * 6的大小，用来存放fd_set_bits fds;
    53. 	bits = select_bits_alloc(size);
    54. 	
    55. 	if (!bits)
    56. 		goto out_nofds;
    57.  
    58. 	// 推指针
    59. 	fds.in      = (unsigned long *)  bits;
    60. 	fds.out     = (unsigned long *) (bits +   size);
    61. 	fds.ex      = (unsigned long *) (bits + 2*size);
    62. 	fds.res_in  = (unsigned long *) (bits + 3*size);
    63. 	fds.res_out = (unsigned long *) (bits + 4*size);
    64. 	fds.res_ex  = (unsigned long *) (bits + 5*size);
    65.  
    66. 	// 把用户态的数据拷贝到内核态，因为用户态的数据不可信。
    67. 	if ((ret = get_fd_set(n, inp, fds.in)) ||
    68. 	    (ret = get_fd_set(n, outp, fds.out)) ||
    69. 	    (ret = get_fd_set(n, exp, fds.ex)))
    70. 		goto out;
    71.  
    72.     // 清空返回给用户态的位图
    73. 	zero_fd_set(n, fds.res_in);
    74. 	zero_fd_set(n, fds.res_out);
    75. 	zero_fd_set(n, fds.res_ex);
    76.  
    77.     // 调用do_select执行具体逻辑
    78. 	ret = do_select(n, &fds, &timeout);
    79.  
    80. 	if (tvp && !(current->personality & STICKY_TIMEOUTS)) {
    81. 		time_t sec = 0, usec = 0;
    82. 		if (timeout) {
    83. 			sec = timeout / HZ;
    84. 			usec = timeout % HZ;
    85. 			usec *= (1000000/HZ);
    86. 		}
    87. 		put_user(sec, &tvp->tv_sec);
    88. 		put_user(usec, &tvp->tv_usec);
    89. 	}
    90.  
    91. 	if (ret < 0)
    92. 		goto out;
    93. 	if (!ret) {
    94. 		ret = -ERESTARTNOHAND;
    95. 		if (signal_pending(current))
    96. 			goto out;
    97. 		ret = 0;
    98. 	}
    99.  
    100.     // 把内核态返回数据的内容拷贝到用户态.
    101. 	set_fd_set(n, inp, fds.res_in);
    102. 	set_fd_set(n, outp, fds.res_out);
    103. 	set_fd_set(n, exp, fds.res_ex);
    104.  
    105. out:
    106. 	select_bits_free(bits, size);
    107. out_nofds:
    108. 	return ret;
    109. }

    方法参数：

            // n是事件数量
            // inp是输入事件的位图
            // outp是输出事件的位图
            // exp是不感兴趣事件的位图（如果是不感兴趣的事件，就会被忽略）
            // tvp是计时器

    fd_set __user：

    fd的位图形式，用一个bit位来代表一个fd，这样非常非常节省空间。__user就代表是用户态传来的数据.

    这段代码并不复杂，这里根据事件的数量，得出内核要开辟多少个位图大小来存放用户态传过来的数据，大部分的操作都是把用户态的内容拷贝到内核态，再把内核态的数据拷贝到用户态。（为什么要把用户态的数据给拷贝到内核态呢？思考一个问题，这里是传入的指针，指针具体指向的数据在用户态，而用户态可能会修改数据，导致数据不安全，所以，对于内核态来说，用户态的数据不可信，所以就需要进行拷贝）

    do_select:

    把用户态的数据进行拷贝以后，进入到do_select方法中进行具体的处理。

    1. int do_select(int n, fd_set_bits *fds, long *timeout)
    2. {
    3. 	struct poll_wqueues table;
    4. 	poll_table *wait;
    5. 	int retval, i;
    6. 	long __timeout = *timeout;
    7.  
    8.  	spin_lock(¤t->files->file_lock);
    9. 	retval = max_select_fd(n, fds);
    10. 	spin_unlock(¤t->files->file_lock);
    11.  
    12. 	if (retval < 0)
    13. 		return retval;
    14. 	n = retval;
    15.  
    16. 	poll_initwait(&table);
    17. 	wait = &table.pt;
    18. 	if (!__timeout)
    19. 		wait = NULL;
    20. 	retval = 0;
    21. 	for (;;) {
    22. 		unsigned long *rinp, *routp, *rexp, *inp, *outp, *exp;
    23.  
    24. 		set_current_state(TASK_INTERRUPTIBLE);
    25.  
    26. 		inp = fds->in; outp = fds->out; exp = fds->ex;
    27. 		rinp = fds->res_in; routp = fds->res_out; rexp = fds->res_ex;
    28.  
    29. 		for (i = 0; i < n; ++rinp, ++routp, ++rexp) {
    30. 			unsigned long in, out, ex, all_bits, bit = 1, mask, j;
    31. 			unsigned long res_in = 0, res_out = 0, res_ex = 0;
    32. 			struct file_operations *f_op = NULL;
    33. 			struct file *file = NULL;
    34.  
    35. 			// 这里是先++ 再*
    36. 			// 取数组的下一位。
    37. 			in = *inp++; out = *outp++; ex = *exp++;
    38.  
    39. 			// 二进制组合，最后还是为0，就代表这三个值都为0
    40. 			all_bits = in | out | ex;
    41. 			if (all_bits == 0) {
    42. 				i += __NFDBITS;
    43. 				continue;
    44. 			}
    45.  
    46. 			for (j = 0; j < __NFDBITS; ++j, ++i, bit <<= 1) {
    47.  
    48. 				// 这个退出条件，代表所有的n都已经遍历完了。
    49. 				if (i >= n)
    50. 					break;
    51. 				if (!(bit & all_bits))
    52. 					continue;
    53.  
    54. 				// 获取到文件
    55. 				file = fget(i);
    56. 				
    57. 				if (file) {
    58. 					f_op = file->f_op;
    59. 					mask = DEFAULT_POLLMASK;
    60. 					if (f_op && f_op->poll)
    61. 						//  这里面会把当前fd和当前current做绑定放入到一个队列中等待
    62. 						mask = (*f_op->poll)(file, retval ? NULL : wait);
    63. 					fput(file);
    64.  
    65. 					// in 必须有bit这一位？
    66. 					if ((mask & POLLIN_SET) && (in & bit)) {
    67. 						res_in |= bit;
    68. 						retval++;
    69. 					}
    70.  
    71. 					// out 必须有bit这一位？
    72. 					if ((mask & POLLOUT_SET) && (out & bit)) {
    73. 						res_out |= bit;
    74. 						retval++;
    75. 					}
    76.  
    77. 					// ex 必须有bit这一位？
    78. 					if ((mask & POLLEX_SET) && (ex & bit)) {
    79. 						res_ex |= bit;
    80. 						retval++;
    81. 					}
    82. 				}
    83. 			}
    84. 			if (res_in)
    85. 				*rinp = res_in;
    86. 			if (res_out)
    87. 				*routp = res_out;
    88. 			if (res_ex)
    89. 				*rexp = res_ex;
    90. 		}
    91. 		// shit
    92. 		wait = NULL;
    93. 		if (retval || !__timeout || signal_pending(current))
    94. 			break;
    95. 		if(table.error) {
    96. 			retval = table.error;
    97. 			break;
    98. 		}
    99. 		__timeout = schedule_timeout(__timeout);
    100. 	}
    101. 	__set_current_state(TASK_RUNNING);
    102.  
    103.  
    104. 	poll_freewait(&table);
    105.  
    106. 	/*
    107. 	 * Up-to-date the caller timeout.
    108. 	 */
    109. 	*timeout = __timeout;
    110. 	return retval;
    111. }

    嗯，代码量确实挺多的，多层for循环嵌套，但是没关系，待笔者细细道来。

    首先，我们得先要明白所有逻辑都在do_select方法中了，所以给事件挂回调的处理、队列的处理都会在这里。

    void poll_initwait(struct poll_wqueues *pwq)
    {
        init_poll_funcptr(&pwq->pt, __pollwait);
        pwq->error = 0;
        pwq->table = NULL;
    }

    函数指针的赋值，把__pollwait函数挂在了poll_wqueues结构体中poll_table pt属性中。

    随之进入到for循环中。

    第一层for循环：都把select称为轮训式多路复用，这里体现的淋漓尽致，因为死循环的机制+schedule_timeout(__timeout)切换上下文的方式（让出CPU的使用权，直到有中断唤醒）

    第二层for循环和第三层for循环：就是来遍历位图，因为位图的一位代表一个fd，所以通过fd获取到file结构体。

    拿到file结构体后，VFS虚拟文件+函数指针的魅力又体现出来了，因为通过if (f_op && f_op->poll) 就可以得知当前的文件系统是否实现了poll函数指针。因为对于poll函数指针而言，普通的文件系统是没有实现的，而socket套接字的文件系统是有实现的。而下文会仔细道来。而poll返回的mask标志位是可以判断是否有读写事件，存在读写事件的话，res_in |= bit就把返回的位图对应置位，并且retval++，后续就会break;退出所有循环，回到sys_select中。

    所以看到mask = (*f_op->poll)(file, retval ? NULL : wait);具体的函数实现。

    poll函数指针socket的具体实现

     

     我们直接看到tcp的实现，但是此方法不过细讲，我们能明白他内部回调了之前poll_initwait方法挂上的__pollwait回调函数即可。

    并且此方法，会返回do_select方法需要的mask标志位。也就是可以理解为，当tcp完成三次握手，建立上连接以后，会根据tcp协议分析出读写标志位（这里网络栈，不过细讲，能明白会建立连接后会得到do_select需要的标志位即可）。

    所以看到__pollwait的实现。

    __pollwait回调函数：

    1. void __pollwait(struct file *filp, wait_queue_head_t *wait_address, poll_table *_p)
    2. {
    3. 	// 骚操作获取到结构体的基址。
    4. 	struct poll_wqueues *p = container_of(_p, struct poll_wqueues, pt);
    5.  
    6. 	// 
    7. 	struct poll_table_page *table = p->table;
    8.  
    9. 	// 队列不存在，或者内部元素为0.
    10. 	// 也就是初始化的操作
    11. 	if (!table || POLL_TABLE_FULL(table)) {
    12. 		struct poll_table_page *new_table;
    13.  
    14. 		new_table = (struct poll_table_page *) __get_free_page(GFP_KERNEL);
    15. 		if (!new_table) {
    16. 			p->error = -ENOMEM;
    17. 			__set_current_state(TASK_RUNNING);
    18. 			return;
    19. 		}
    20. 		new_table->entry = new_table->entries;
    21. 		new_table->next = table;
    22. 		p->table = new_table;
    23. 		table = new_table;
    24. 	}
    25.  
    26. 	/* Add a new entry */
    27. 	// 放入到队列。
    28. 	{
    29. 		struct poll_table_entry * entry = table->entry;
    30.  
    31. 		// 意思是poll_table_entry是一个连续的空间，数组？
    32. 		table->entry = entry+1;
    33.  
    34. 		// 原子性加引用。代表被使用了。
    35. 	 	get_file(filp);
    36. 		
    37. 	 	entry->filp = filp;
    38. 		
    39. 		entry->wait_address = wait_address;
    40.  
    41. 		// 初始化wait_queue_t wait;
    42. 		// 把对应的进程结构体、回调钩子赋值
    43. 		init_waitqueue_entry(&entry->wait, current);
    44.  
    45. 		// wait_address这个是sock维护的队列
    46. 		// &entry->wait这个是队列中的元素
    47. 		// 所以这是放入到队列中
    48. 		add_wait_queue(wait_address,&entry->wait);
    49. 	}
    50. }

    此方法就是添加到等待队列的具体实现。上面是初始化的操作，下面是添加到等待队列的操作。

    看到init_waitqueue_entry方法的实现.

    static inline void init_waitqueue_entry(wait_queue_t *q, struct task_struct *p)
    {
        q->flags = 0;
        q->task = p;
        // 继续放钩子
        // 被唤醒的钩子。
        q->func = default_wake_function;
    }

    这里给等待队列中的元素又挂上了default_wake_function钩子回调函数。

    看到add_wait_queue方法的实现.

    // wait_queue_head_t *q 是当前tcp的等待队列。

    // wait_queue_t * wait 是当前文件对应的等待队列的元素。

    void add_wait_queue(wait_queue_head_t *q, wait_queue_t * wait)
    {
        unsigned long flags;

        wait->flags &= ~WQ_FLAG_EXCLUSIVE;
        spin_lock_irqsave(&q->lock, flags);
        __add_wait_queue(q, wait);
        spin_unlock_irqrestore(&q->lock, flags);
    }

    这里把当前文件对应的等待队列的元素链到tcp的等待队列，而等待队列的元素中的func函数指针是挂上了default_wake_function钩子。

    等待底层网卡给CPU发出中断信号，CPU响应中断，中断处理方法中去处理等待队列。并且会回调default_wake_function钩子（这里的论证太底层，偏硬件，笔者能力有限，笔者只能给出论据，不能给出论证了....）

    default_wake_function回调函数：

    1. int default_wake_function(wait_queue_t *curr, unsigned mode, int sync)
    2. {
    3. 	task_t *p = curr->task;
    4. 	return try_to_wake_up(p, mode, sync);
    5. }

    这里就特别的简单了，把当前进程给唤醒。如何唤醒：把task_struct的state赋值为TASK_RUNNING，然后等待时钟中断的中断处理函数调度此进程。

    然后当前进程苏醒过来以后，又回到了do_select方法的schedule_timeout(__timeout); 中，此时又开始了新的一轮轮询，而这轮轮询是通过poll函数指针获取到mask标志位（因为当前中断唤醒都是因为tcp建立了连接通过default_wake_function把进程给唤醒，所以已经有读写事件的产生）然后通过mask判断把对应的位图置位，然后返回给用户态。

    总结：

    select多路复用的弊端

    1. 文件大小的限制
    2. 多次轮训——内核态的轮训，返回的结果在用户态还需要轮训
    3. 多次数据拷贝，不过是位图占用空间不多。

    最后，如果本帖对您有一定的帮助，希望能点赞+关注+收藏！您的支持是给我最大的动力，后续会一直更新各种框架的使用和框架的源码解读~！

  • 相关阅读:
    Java的选择排序、冒泡排序、插入排序
    网络编程学习part1
    发展之路是怎么样？ 防爆机器人的未来发展之路是怎么样
    python数据分析基础—取某列字符的前几个字符
    心理软件使用问题三
    2023年9月到2024年1月计划安排
    还记得2048怎么玩吗？快来玩会儿(摸鱼)吧
    Android应用开发架构之MVI：原理、流程与实战指南
    spring-boot-starter-data-redis2.X连接redis7
    缓存篇—缓存雪崩、缓存击穿、缓存穿透
  • 原文地址：https://blog.csdn.net/qq_43799161/article/details/126279876