linux实时进程的负载均衡详解

2023-10-23 11:49| 来源: 网络整理| 查看: 265

在 Linux 内核中会为每个 CPU 创建一个可运行进程队列，由于每个 CPU 都拥有一个可运行进程队列，那么就有可能会出现每个可运行进程队列之间的进程数和负载不一样的问题。

另外，考虑系统中各个CPU的算力，让CPU获得和其算力匹配的负载。例如在一个6个小核+2个大核的系统中，整个系统如果有800的负载，那么每个CPU上分配100的负载其实是不均衡的，因为大核CPU可以提供更强的算力。

当负载不均衡的时候，任务需要在CPU之间迁移，不同形态的迁移会有不同的开销。例如一个任务在小核cluster上的CPU之间的迁移所带来的性能开销一定是小于任务从小核cluster的CPU迁移到大核cluster的开销。因此，为了更好的执行负载均衡，我们需要构建和CPU拓扑相关的数据结构，也就是根域（root_domain）、调度域（sched_domain）和调度组（sched_group）的概念。

其中，根域（root_domain）主要是负责实时调度类（包括 dl 和 rt 调度类）负载均衡而设计的数据结构，协助 dl 和 rt 调度类完成实时任务的合理调度。

1、Root Domain

cpuset 提供了一个把 CPU 分成子集被一个进程或者或一组进程使用的机制。每个互斥的 cpuset 定义了一个与其他 cpuset 或 CPU 分离的孤岛域（isolated domain，也叫作 root domain）。

每个 root domian 有关的信息存在 struct root_domain 结构（对象）中。

struct root_domain { atomic_t refcount; // root domain 的引用计数，当 root domain 被运行队列引用时加一，反之减一。 atomic_t rto_count; // 过载的（overload）的 CPU 的数目。 struct rcu_head rcu; cpumask_var_t span; //属于该 root domain 的运行队列的可用 CPU的范围，cpumask_var_t掩码。 cpumask_var_t online; int overload; //表明该 root domain 有任一 CPU 有多于一个的可运行任务。 /* Indicate one or more cpus over-utilized (tipping point) */ int overutilized; /* * The bit corresponding to a CPU gets set here if such CPU has more * than one runnable -deadline task (as it is below for RT tasks). */ cpumask_var_t dlo_mask; atomic_t dlo_count; struct dl_bw dl_bw; struct cpudl cpudl; #ifdef HAVE_RT_PUSH_IPI /* * For IPI pull requests, loop across the rto_mask. */ struct irq_work rto_push_work; raw_spinlock_t rto_lock; /* These are only updated and read within rto_lock */ int rto_loop; int rto_cpu; /* These atomics are updated outside of a lock */ atomic_t rto_loop_next; atomic_t rto_loop_start; #endif /* * The "RT overload" flag: it gets set if a CPU has more than * one runnable RT task. */ cpumask_var_t rto_mask; // 某 CPU 有多于一个的可运行实时任务，对应的位被设置，cpumask_var_t掩码。 struct cpupri cpupri; //包含在 root domain 中的CPU 优先级管理结构成员 unsigned long max_cpu_capacity; /* * NULL-terminated list of performance domains intersecting with the * CPUs of the rd. Protected by RCU. */ struct perf_domain __rcu *pd; };

无论何时一个互斥 cpuset 被创建，一个新 root domain 对象也会被创建，信息来自 CPU 成员。所有的实时调度决定只在一个 root domain 的范围内作出决定。一般的嵌入式设备，只有一个root domain。

2、CPU优先级管理

CPU 优先级管理（CPU Priority Management）跟踪系统中每个 CPU 的优先级，为了让进程迁移的决定更有效率。

CPU 优先级有 102 个:

cpupritask prioCPUPRI_INVALID (-1)-1CPUPRI_IDLE (0)MAX_PRIO (140)CPUPRI_NORMAL (1)MAX_RT_PRIO ~ MAX_PRIO-1 (100~139)2~10199~0

task prio转cpupri的函数如下：

/* Convert between a 140 based task->prio, and our 102 based cpupri */ static int convert_prio(int prio) { int cpupri; if (prio == CPUPRI_INVALID) cpupri = CPUPRI_INVALID; else if (prio == MAX_PRIO) cpupri = CPUPRI_IDLE; else if (prio >= MAX_RT_PRIO) cpupri = CPUPRI_NORMAL; else cpupri = MAX_RT_PRIO - prio + 1; return cpupri; } cpupri 数值越大表示优先级越高（用的是减法）。处于CPUPRI_INVALID状态的 CPU 没有资格参与 task routing。 cpupri 属于 root domain level 的范围。每个互斥的 cpuset 由一个含有 cpupri 数据的 root momain 组成。系统从两个维度来维护这些 CPU 状态： CPU 的优先级，由任务优先级映射而来在某个优先级上的 CPU掩码，用来查找某一优先级上CPU的分布情况

通过cpupri_find()和cpupri_set()来查找和设置 CPU 优先级是实时负载均衡快速找到要迁移的任务的关键。

struct cpupri_vec { atomic_t count; // 在这个优先级上的 CPU 的数量 cpumask_var_t mask; // 在这个优先级上的 CPU 位掩码 }; struct cpupri { struct cpupri_vec pri_to_cpu[CPUPRI_NR_PRIORITIES]; // 持有关于一个 cpuset 在某个特定的优先级上的所有 CPU 的信息。 int *cpu_to_pri; // 指示一个 CPU 的优先级，注意与任务的优先级区分。 }; 2.1 cpupri_find()

cpupri_find() 查找系统（root domain）中最佳（优先级最低）的 CPU，即可以将任务分配给它。成功找到返回 1，找不到返回 0，结果通过lowest_mask返回。

int cpupri_find(struct cpupri *cp, struct task_struct *p, struct cpumask *lowest_mask) { int idx = 0; //将任务的有效优先级转为 CPU 优先级 int task_pri = convert_prio(p->prio); BUG_ON(task_pri >= CPUPRI_NR_PRIORITIES); //从最低优先级开始检查，因为跑着低优先级任务的 CPU 更适合被 routing //CPUPRI_IDLE (0)、CPUPRI_NORMAL (1) for (idx = 0; idx < task_pri; idx++) { //优先级为idx的所有 CPU 的信息（count与mask） struct cpupri_vec *vec = &cp->pri_to_cpu[idx]; int skip = 0; //优先级为idx的CPU个数为0 if (!atomic_read(&(vec)->count)) skip = 1; /*当访问向量的时候，我们需要先读计数器，做内存屏障，然后读掩码。注意：这里仍然是有竞争的，但我们可以处理它。理想的情况下，我们只想看那些被设置的掩码。如果一个掩码没有被设置，那么唯一错误的事情是，我们多做了一些没必要的工作。如果由于内存屏障的缘故，我们读到一个零计数，但掩码却被设置了，那只能发生在当运行队列的最高优先级的任务离开运行队列的时候，在这种情况下，它会紧跟着一个拉操作。如果我们正在处理的任务没能找到一个合适的去处，且运行队列正运行在一个更低的优先级，那么拉请求将会把这个任务拉过来。*/ smp_rmb(); /* Need to do the rmb for every iteration */ if (skip) continue; //任务亲和的 CPU 掩码与向量的掩码 “与” 运算，看有没有合适的 CPU if (cpumask_any_and(p->cpus_ptr, vec->mask) >= nr_cpu_ids) continue; if (lowest_mask) { //任务亲和的 CPU 掩码与向量的掩码 “与” 运算赋给 lowest_mask cpumask_and(lowest_mask, p->cpus_ptr, vec->mask); //find the first set bit in a memory region，Returns >= nr_cpu_ids if no cpus set. //lowest_mask为空，cpumask_any()返回结果 >= nr_cpu_ids，表示该轮查找不成功，进行下一次循环 if (cpumask_any(lowest_mask) >= nr_cpu_ids) continue; } return 1; } return 0; } 2.2 cpupri_set()

设置 CPU 的优先级为newpri（进程优先级）对应的CPU优先级

void cpupri_set(struct cpupri *cp, int cpu, int newpri) { // 根据 CPU 找到 CPU 的优先级 int *currpri = &cp->cpu_to_pri[cpu]; int oldpri = *currpri; int do_mb = 0; //进程优先级转为 CPU 优先级 newpri = convert_prio(newpri); BUG_ON(newpri >= CPUPRI_NR_PRIORITIES); //CPU 优先级未改变，直接返回 if (newpri == oldpri) return; //如果 cpu 当前被映射到一个不同的值，我们需要把它映射到 pri_to_cpu 优先级向量里的掩码 //对应的位，然后把旧向量里的掩码位清除。注意，我们必须先添加新的值，否则我们会有在 //cpupri_find 循环优先级时错过该 cpu 的风险 if (likely(newpri != CPUPRI_INVALID)) { struct cpupri_vec *vec = &cp->pri_to_cpu[newpri]; //设置向量的 cpu 位掩码 cpumask_set_cpu(cpu, vec->mask); //当新增加一个向量的时候，我们先更新掩码，做一个写屏障，然后再更新计数， //这样确保当计数被设置的时候该向量是可见的 smp_mb__before_atomic(); atomic_inc(&(vec)->count); do_mb = 1; } //旧优先级的值是有效值 if (likely(oldpri != CPUPRI_INVALID)) { //取将要移除的优先级向量 struct cpupri_vec *vec = &cp->pri_to_cpu[oldpri]; //因为修改 vec->count 的顺序是重要的，我们必须确保新优先级的更新在我们减小旧优先级前被看到。 //这确保当我们提高运行队列的优先级的时候，循环能看到一个或其他的优先级。 //当我们降低优先级的时候我们并不关心，因为那总会触发一次实时拉操作。 //如果我们更新新的优先级向量，我们只需要做一个内存屏障 if (do_mb) smp_mb__after_atomic(); //当从向量移除的时候，我们先减小计数器，再做一个内存屏障，然后才清除位掩码 atomic_dec(&(vec)->count); smp_mb__after_atomic(); //清除向量的 cpu 位掩码 cpumask_clear_cpu(cpu, vec->mask); } //最后才是 cpu 优先级的值的更新 *currpri = newpri; } 3、RT负载均衡的时机

RT负载均衡是通过push_rt_task()与pull_rt_task()函数实现的。其中，push_rt_task()会在以下时间点被调用：

直接调用：task_woken_rt() 中调用 push_rt_tasks()，条件比较苛刻，如下：表示为唤醒的任务p不是rq上正在running的任务，且当前rq也没有设置resched标志位(不会马上重新调度)，且p也允许在其它CPU上运行，且rq当前正在运行的任务是DL或RT任务，且rq的当前任务只能在当前CPU运行或优先级比p更高。间接调用（通过balance_callback回调）：进程调度类发生改变时（比如通过sched_setscheduler系统调用改变调度类，通过rt_mutex_setprio进行优先级继承）；间接调用（通过balance_callback回调）：发生调度时（这时候可能有一个实时进程被更高优先级的实时进程抢占了）；

pull_rt_task()会在以下时间点被调用：

直接调用：pick_next_task_rt() 选择下一个要调度的实时进程之前，先balance一下，当前运行队列的最高优先级比prev任务的优先级低，这个时候拉一把；间接调用（通过balance_callback回调）：进程调度类发生改变时（比如通过sched_setscheduler系统调用改变调度类，通过rt_mutex_setprio进行优先级继承）；间接调用（通过balance_callback回调）：进程优先级发生变化时（比如通过sched_setscheduler系统调用改变优先级，通过rt_mutex_setprio进行优先级继承）；

详细过程如下：

www.processon.com/view/link/6…

RT实时进程负载均衡.png

4、PUSH 推任务迁移 4.1 PUSH 推任务的基本思想

PUSH 推任务迁移是指一个CPU将本运行队列上最高优先级（不在运行）的实时任务，转移到另一个运行队列的操作。

具体步骤如下：

找到本CPU运行队列上优先级最高的不在运行的可运行实时任务TaskA；搜索一个优先级更低的队列，就是该运行队列上当前正在运行的任务可以被TaskA抢占的队列；如何搜索一个优先级更低的队列，上述CPU 优先级管理结构就是用于找到一个有最低优先级运行队列的 CPU 掩码，从所有的候选者中选择唯一的最佳CPU。首先，把TaskA给上一次最后执行TaskA的 CPU，由于它的 cache 很可能还是热的；其次，当前执行推任务TaskA的 CPU 如果在找到的最低优先级 CPU 掩码里，且在同一调度域的分支上，则被选中。否则，这说明当前执行推任务TaskA的 CPU 不在同一调度域的分支上，那么在最低优先级CPU掩码与调度域的CPU span的交集中选第一个 CPU 作为最佳 CPU。如果在调度域里没有匹配最低优先级 CPU 掩码的 CPU，那么就把这个在最低优先级 CPU 掩码里，但不在同一调度域上的执行推任务的当前 CPU 返回。如果也失败了，从掩码中随机选择一个 CPU。 4.2 PUSH 推任务的时机

push_rt_task()函数会在以下时间点被调用（实时进程要保证实时性而不是负载功耗的均衡）：

非正在运行的普通进程变成实时进程时（比如通过sched_setscheduler系统调用）；发生调度之后（这时候可能有一个实时进程被更高优先级的实时进程抢占了）；实时进程被唤醒之后，如果不能马上在当前 CPU 上运行（它不是当前 CPU 上优先级最高的进程，可能是其他CPU上的最高优先级）； 4.3 PUSH 推任务的实现 static void push_rt_tasks(struct rq *rq) { /*push_rt_task()如果移动了一个实时任务将会返回 1。这个循环会一直执行到给定队列没有任务可以推走*/ while (push_rt_task(rq)) ; } /*如果当前 CPU 有超过一个实时任务，看有没有非运行态任务可以被迁移到另一个运行着优先级低一些的任务的 CPU*/ static int push_rt_task(struct rq *rq) { struct task_struct *next_task; struct rq *lowest_rq; int ret = 0; /*如果该队列没有过载，即没有一个可以被迁移的实时任务，则没有实时任务可以被推走*/ if (!rq->rt.overloaded) return 0; /*首先在该队列的可推链表里找有没有合适的任务，如果没有则返回。之前任务入列的时候会通过 enqueue_pushable_task() 把可能的任务加进可推队列。*/ next_task = pick_next_pushable_task(rq); if (!next_task) return 0; retry: if (unlikely(next_task == rq->curr)) { /*选出的任务竟然是 rq 的当前任务*/ WARN_ON(1); return 0; /*这种情况不是期望的行为，但也没可以推走的任务*/ } /*从可推队列里选出的任务可能是一个刚溜进的来的高优先级任务，如果是这种情况，重新调度当前任务即可，也不需要推给别人了。*/ if (unlikely(next_task->prio < rq->curr->prio)) { resched_curr(rq); return 0; } /* We might release rq lock */ get_task_struct(next_task); /*在选出来的任务优先级低的队列中，找到优先级最低的队列。如果找到的话，当前队列 rq 和最低优先级队列 lowest_rq 都会处于锁定状态*/ lowest_rq = find_lock_lowest_rq(next_task, rq); if (!lowest_rq) { /*如果没找到优先级最低的队列*/ struct task_struct *task; /*因为 find_lock_lowest_rq() 会释放 rq->lock 锁，所以之前选中的 next_task 有已经被迁移的可能。*/ task = pick_next_pushable_task(rq); if (task_cpu(next_task) == rq->cpu && task == next_task) { /*next_task 的 CPU 仍然是当前队列的 CPU，还没有被迁移，且仍然是下一个具备被迁移资格的任务，但我们找不到一个合适的运行队列来接纳它。这种情况不需要重试，直到其他 CPU 在合适的时候拉走它。*/ goto out; } if (!task) /*找不到优先级最低的队列，也没有其他任务可以推，直接退出*/ goto out; /*找不到优先级最低的队列，且之前选中的 next_task 已经被迁移走了，那么把刚选出的 task 作为新的 next_task 进行下一次重试*/ put_task_struct(next_task); /*对应到之前的 get_task_struct()*/ next_task = task; goto retry; } /*如果找到优先级最低的队列，开始迁移 next_task*/ /*选中的任务从当前队列 rq 中出列*/ deactivate_task(rq, next_task, 0); /*选中的任务 cpu 设置为将要迁移至的最低优先级队列的 cpu*/ set_task_cpu(next_task, lowest_rq->cpu); /*选中的任务进入最低优先级队列*/ activate_task(lowest_rq, next_task, 0); ret = 1; /*返回值设置为 1 表示成功*/ /*把最低优先级队列的当前任务设置为“需要被立即重新调度”*/ resched_curr(lowest_rq); /*同时解锁当前队列 rq 和最低优先级队列 lowest_rq*/ double_unlock_balance(rq, lowest_rq); out: put_task_struct(next_task); /*对应到之前的 get_task_struct()*/ return ret; } /*返回值为指向运行队列的指针 - *rq，找不到时返回 NULL 该函数的主要工作： 1. 调用 find_lowest_rq() 去查找 2. 解决好锁的问题 3. 尝试 3 次 */ static struct rq *find_lock_lowest_rq(struct task_struct *task, struct rq *rq) { struct rq *lowest_rq = NULL; /*指向要返回的优先级最低队列*/ int tries; int cpu; /*目前只尝试三次*/ for (tries = 0; tries < RT_MAX_TRIES; tries++) { cpu = find_lowest_rq(task); /*如果找不到一个合适的队列，或者找到的队列就是当前队列，则跳出循环，用已有的 lowest_rq 的值。*/ if ((cpu == -1) || (cpu == rq->cpu)) break; /*否则认为找到了优先级最低队列，更新 lowest_rq 的值为该 CPU 所属 rq*/ lowest_rq = cpu_rq(cpu); if (lowest_rq->rt.highest_prio.curr prio) { /*如果目标运行队列所记录的最高优先级高于或等于要推走的任务，重试不释放任何锁，因此不可能导致不同的结果。所以这里设置为无法找到队列就跳出循环了。记住，highest_prio记录的是该队列上的最高优先级，但不一定是该队列上正在运行的任务，有可能是该加入队列的还未被调度的高优先级任务，因此是有可能与 CPU 优先级的记录不一致的。*/ lowest_rq = NULL; break; } /*因为要在当前队列和目标队列之间迁移，因此需要同时锁定这两个队列*/ if (double_lock_balance(rq, lowest_rq)) { /*double_lock_balance()会有个先解锁再加锁的过程，这个间隙有可能会发生一些改变，因此需要重新做以下检查： - 之前从 rq 里选出来的 task 现在其 rq 域不再指向 rq 了 - 进程的亲和性被修改导致不再在最低优先级队列的 CPU 里了 - 当前队列正在运行的任务就是该task - 任务的调度器不再是实时调度器 - 任务不在运行队列上了以上任何一种情况出现都被认为查找失败。*/ if (unlikely(task_rq(task) != rq || !cpumask_test_cpu(lowest_rq->cpu, tsk_cpus_allowed(task)) || task_running(rq, task) || !rt_task(task) || !task_on_rq_queued(task))) { double_unlock_balance(rq, lowest_rq); lowest_rq = NULL; break; } } /*如果能锁定成功且能通过以上一系列检查，这时才认为该运行队列是合适被推给任务的，可以跳出循环了，此时返回值 lowest_rq 具备有效值。解除 double_unlock_balance(rq, lowest_rq) 的地方在 push_rt_task()，因为在这期间要一直锁定着两个队列。*/ if (lowest_rq->rt.highest_prio.curr > task->prio) break; /*否则进行重试*/ double_unlock_balance(rq, lowest_rq); lowest_rq = NULL; } return lowest_rq; } static struct task_struct *pick_next_pushable_task(struct rq *rq) { struct task_struct *p; if (!has_pushable_tasks(rq)) return NULL; /*plist是按照优先级由高到底进行排序的，所以 first entry 是链表里优先级最高的*/ p = plist_first_entry(&rq->rt.pushable_tasks, struct task_struct, pushable_tasks); /*返回前检查一些不该存在的错误*/ BUG_ON(rq->cpu != task_cpu(p)); /*任务 CPU 与队列 CPU 不是同一个*/ BUG_ON(task_current(rq, p)); /*选出的任务竟然是当前任务*/ BUG_ON(tsk_nr_cpus_allowed(p) rd->cpupri, task, lowest_mask)) return -1; /* No targets found */ /*此时，我们建立了一个系统中运行最低优先级任务的 CPU 的掩码。现在我们想根据我们的亲和性和拓扑结构选出最佳的那个。我们优先选择上一次执行任务的 CPU，因为它的 cache 很可能还是热的。*/ if (cpumask_test_cpu(cpu, lowest_mask)) /*测试 task 的 CPU 是否在掩码中*/ return cpu; /*如果在掩码中，优先选择该 CPU*/ /*否则，我们根据调度域的范围映射来找出那个 CPU 在逻辑上最接近我们的热缓存数据。*/ if (!cpumask_test_cpu(this_cpu, lowest_mask)) this_cpu = -1; /* Skip this_cpu opt if not among lowest */ /*如果当前 CPU 不在掩码给出的 CPU 里，标记成 -1 跳过它。*/ rcu_read_lock(); /*domain tree 有 RCU quiescent state transition 的保护*/ for_each_domain(cpu, sd) { /*由下至上遍历调度域，优先选择邻近的*/ if (sd->flags & SD_WAKE_AFFINE) { int best_cpu; /*抢占 "this_cpu" 比抢占一个远程的处理器开销更低。当前 CPU 如果在之前的最低优先级掩码里，且在同一调度域的分支上，则有可能被选中，原因如上所述*/ if (this_cpu != -1 && cpumask_test_cpu(this_cpu, sched_domain_span(sd))) { rcu_read_unlock(); return this_cpu; } /*如果当前 CPU 不在同一调度域的分支上，在 "最低优先级掩码" 与 "调度域的 CPU span" 的交集中选第一个 CPU 作为最佳 CPU*/ best_cpu = cpumask_first_and(lowest_mask, sched_domain_span(sd)); if (best_cpu < nr_cpu_ids) { rcu_read_unlock(); return best_cpu; } } } rcu_read_unlock(); /*最后，如果在调度域里没有匹配 lowest_mask 的 CPU，那么就把这个在 lowest_mask 掩码里，但不在同一调度域分支上的当前 CPU 返回。这意味此时当前 CPU （就调度域划分而言）是远程的。*/ if (this_cpu != -1) return this_cpu; /*如果当前 CPU 既不在 lowest_mask 掩码里，也不在同一调度域分支上。从最低优先级掩码给定的 CPU 中随机选择一个 CPU 返回。*/ cpu = cpumask_any(lowest_mask); if (cpu < nr_cpu_ids) return cpu; return -1; /*这样的 CPU 如果是非法值，那只能说找不到了。*/ } 5、PULL 拉任务迁移 5.1 PULL 拉任务的基本思想 pull_rt_task()算法着眼于一个 root domain 中所有过载的运行队列，检查它们是否有一个实时任务能运行在本CPU的运行队列（本CPU 在任务的task->cpus_allowed_mask 中）且其优先级高于本CPU上将要被调度的任务。扫描 root domain 中所有过载的运行队列之后终结。因此，PULL 拉操作可能拉多于一个任务到本CPU运行队列。 5.2 运行队列过载标志在进程进入rt_rq时会调用inc_rt_migration() __enqueue_rt_entity() -> inc_rt_tasks() -> inc_rt_prio() -> inc_rt_migration() -> update_rt_migration() -> inc_rt_group() 在进程移出rt_rq时会调用dec_rt_migration() __dequeue_rt_entity() -> dec_rt_tasks() -> dec_rt_prio() -> dec_rt_migration() -> update_rt_migration() -> dec_rt_group()

其中：

rt_nr_total：增加或减小该实时运行队列rt_rq的实时任务数 rt_nr_migratory：如果该任务允许在多于一个 CPU 上运行，增加或减小该实时运行队列rt_rq的可迁移的实时任务数 overloaded：调用共用函数update_rt_migration()更新队列过载标志 rt_nr_running在inc_rt_tasks()或dec_rt_tasks()时更新队列过载标志 static inline void rt_set_overload(struct rq *rq) { if (!rq->online) return; /*把运行队列所在 CPU 加入到运行队列所在 root_domain 上过载的 CPU 掩码*/ cpumask_set_cpu(rq->cpu, rq->rd->rto_mask); smp_wmb(); /*增加运行队列所在 root_domain 上过载 CPU 的计数*/ atomic_inc(&rq->rd->rto_count); } static inline void rt_clear_overload(struct rq *rq) { if (!rq->online) return; /*减小运行队列所在 root_domain 上过载 CPU 的计数*/ atomic_dec(&rq->rd->rto_count); /*清除对应的过载的 CPU 的掩码*/ cpumask_clear_cpu(rq->cpu, rq->rd->rto_mask); } static void update_rt_migration(struct rt_rq *rt_rq) { /*如果有可迁移实时进程，且实时进程数量多于一个*/ if (rt_rq->rt_nr_migratory && rt_rq->rt_nr_total > 1) { if (!rt_rq->overloaded) { /*如果过载标志还没设置，需要做相应的设置*/ rt_set_overload(rq_of_rt_rq(rt_rq)); rt_rq->overloaded = 1; } } else if (rt_rq->overloaded) { /*如果上一个条件未满足，且过载标志设成已过载，需要清理*/ rt_clear_overload(rq_of_rt_rq(rt_rq)); rt_rq->overloaded = 0; } } static inline int rt_overloaded(struct rq *rq) { /*返回运行队列所在 root_domain 上过载的 CPU 数*/ return atomic_read(&rq->rd->rto_count); } 5.3 PULL 拉任务的实现 static void pull_rt_task(struct rq *this_rq) { int this_cpu = this_rq->cpu, cpu; bool resched = false; struct task_struct *p; struct rq *src_rq; /*如果当前队列的 root_domain 的 rto_count 不为 0，说明已经过载了，无需拉任务进来*/ if (likely(!rt_overloaded(this_rq))) return; smp_rmb(); /*这个 feature 容我们稍后阐述，很多情况下不会开启这个 feature*/ #ifdef HAVE_RT_PUSH_IPI if (sched_feat(RT_PUSH_IPI)) { tell_cpu_to_push(this_rq); /*告诉过载的 CPU 将任务推给我们（this_rq）*/ return; } #endif /*逐个遍历当前队列所属 root_domain 的实时过载 CPU 掩码上的 CPU*/ for_each_cpu(cpu, this_rq->rd->rto_mask) { if (this_cpu == cpu) /*如果是当前 CPU，则跳过，因为不用把自己队列上的任务拉给自己*/ continue; /*记录这次迭代拉任务的源运行队列*/ src_rq = cpu_rq(cpu); /*如果源运行队列的可推送任务链表上的最高优先级任务优先级低于或等于当前队列的最高优先级任务优先级，跳过该源队列。源运行队列的已排队最高优先级任务马上有机会被调度，不需要拉过来。因为它比当前队列排第一的任务优先级还低，拉过来也是要等。注意这里取源运行队列数据的时候没有占用 src_rq->lock 的锁，原因见上面注释。主要的意思是，即便是在临界区时优先级变高了，源队列会推过来，而不是拉。如果变低了，那也不用拉。*/ if (src_rq->rt.highest_prio.next >= this_rq->rt.highest_prio.curr) continue; /*double_lock_balance()潜在地会丢失 this_rq 的锁，其他 CPU 有可能趁此修改了 this_rq*/ double_lock_balance(this_rq, src_rq); /*我们一次只拉一个任务，该任务在它的运行队列（源队列）上是可推的，没有其他的了。注意，第二个参数是 this_cpu*/ p = pick_highest_pushable_task(src_rq, this_cpu); /*如果将要拉过来的任务优先级比当前队列的最高优先级高。这里需要再次比较是因为 double_lock_balance()会先释放 this_rq 的锁再加锁，因此有变化的可能*/ if (p && (p->prio < this_rq->rt.highest_prio.curr)) { WARN_ON(p == src_rq->curr); /*将要拉的任务是源队列上当前运行的任务，警告*/ WARN_ON(!task_on_rq_queued(p)); /*任务的状态不是“已排队”，警告*/ /*有一个 p 的优先级比它在的 CPU 上正在运行的任务的优先级高的机会。这就是 p 正好被唤醒但还没有来得及被调度到。我们只在 p 的优先级比它当前运行队列上正在运行的任务优先级低的时候去拉。因为马上有机会被调度的进程没必要拉，所以这里 skip 该进程*/ if (p->prio < src_rq->curr->prio) goto skip; /*否则，开始下面的迁移操作*/ resched = true; /*设置函数本地的重调度标志*/ deactivate_task(src_rq, p, 0); /*从源队列摘下任务*/ set_task_cpu(p, this_cpu); /*设置任务 CPU 为本 CPU*/ activate_task(this_rq, p, 0); /*任务放到本队列*/ /*注意，这里没有 break 而是继续遍历当前 root_domain 的实时过载 CPU 掩码上的 CPU*/ /*我们继续搜索，以防万一在其他的运行队列上有更高优先级的任务。（可能性低，但是是可能的）*/ } skip: double_unlock_balance(this_rq, src_rq); /*在此次迭代的最后解锁*/ } /*如果迭代过程中拉到了任务，毫无疑问要重新调度*/ if (resched) resched_curr(this_rq); } static struct task_struct *pick_highest_pushable_task(struct rq *rq, int cpu) { struct plist_head *head = &rq->rt.pushable_tasks; struct task_struct *p; /*如果运行队列的实时运行队列的可推任务链表为空，返回 NULL*/ if (!has_pushable_tasks(rq)) return NULL; /*遍历可推任务链表上的任务，选出优先级最高的任务返回。因为可推任务链表是按优先级顺序排序的，因此排在前面的任务优先级更高*/ plist_for_each_entry(p, head, pushable_tasks) { if (pick_rt_task(rq, p, cpu)) return p; } return NULL; } static int pick_rt_task(struct rq *rq, struct task_struct *p, int cpu) { /*选中的任务不能是正在运行的任务，而且还允许在目标 CPU 上运行*/ if (!task_running(rq, p) && cpumask_test_cpu(cpu, tsk_cpus_allowed(p))) return 1; return 0; } 6、RT_PUSH_IPI

RT_PUSH_IPI是用 IPI 的方式触发实时任务的推迁移来代替拉迁移，因为多个 CPU 因为优先级降低都同时向一个 CPU 上的运行队列拉任务，导致大量竞争该队列的锁rq lock，导致这多个CPU同时阻塞；反之，如果只触发调度器 IPI 给该 CPU，让该队列自己推一个任务出来，就不会有大量锁争用的问题，同一时间只有push的那个CPU得到锁。

关于IPI的介绍可以参考： 4.2 IPI处理器间中断

RT_PUSH_IPI的开启关闭的接口：

#ifdef HAVE_RT_PUSH_IPI /*为了避免多个 CPU 同时降低其优先级的惊群攻击（回忆之前所说的，进程优先级降低的时候会引发拉迁移），且有一个 CPU 上有一个实时任务可以被迁移并等待运行，这样其他的 CPU 会尽可能尝试获取那个 CPU 的 `rq lock` 从而造成一次大型的争用，在这种场景下更好的方案是，发送一个 IPI 给那个 CPU，让它把那个实时任务推到它该去的地方*/ SCHED_FEAT(RT_PUSH_IPI, true) #endif

开启或关闭该功能：

# mount -t debugfs nodev /sys/kernel/debug # echo RT_PUSH_IPI > /sys/kernel/debug/sched_features # echo NO_RT_PUSH_IPI > /sys/kernel/debug/sched_features 6.1 RT_PUSH_IPI的实现 static void pull_rt_task(struct rq *this_rq) { int this_cpu = this_rq->cpu, cpu; ...... /*这个 feature 容我们稍后阐述，很多情况下不会开启这个 feature*/ #ifdef HAVE_RT_PUSH_IPI if (sched_feat(RT_PUSH_IPI)) { tell_cpu_to_push(this_rq); /*告诉过载的 CPU 将任务推给我们（this_rq）*/ return; } #endif #define RT_PUSH_IPI_EXECUTING 1 #define RT_PUSH_IPI_RESTART 2 static void tell_cpu_to_push(struct rq *rq) { int cpu; /*如果在 IPI 遍历过程中源 CPU 再次降低优先级，会进入以下条件*/ if (rq->rt.push_flags & RT_PUSH_IPI_EXECUTING) { raw_spin_lock(&rq->rt.push_lock); /* Make sure it's still executing */ if (rq->rt.push_flags & RT_PUSH_IPI_EXECUTING) { rq->rt.push_flags |= RT_PUSH_IPI_RESTART; /*设置重新遍历标志*/ raw_spin_unlock(&rq->rt.push_lock); return; } raw_spin_unlock(&rq->rt.push_lock); } /*能走到这儿，说明还没有进行 IPI 遍历。 push CPU 检查的开始为当前队列 CPU（它会被跳过），既是遍历的起点，也是遍历的终点*/ rq->rt.push_cpu = rq->cpu; cpu = find_next_push_cpu(rq); if (cpu >= nr_cpu_ids) /*搜索结束标志为 nr_cpu_ids*/ return; /*搜索未结束，或者一次搜索开始了，变更 push_flags*/ rq->rt.push_flags = RT_PUSH_IPI_EXECUTING; /*把找到的 cpu 排入 irq 工作队列，工作内容为 rq->rt.push_work 指向的函数。这是一个异步的过程，目的是让找到的 cpu 把任务推到这里来。*/ irq_work_queue_on(&rq->rt.push_work, cpu); } static int find_next_push_cpu(struct rq *rq) { struct rq *next_rq; int cpu; while (1) { cpu = rto_next_cpu(rq);/*得到本次要检查的 cpu，该函数能处理回绕的情况*/ if (cpu >= nr_cpu_ids) /*搜索结束标志为 nr_cpu_ids*/ break; next_rq = cpu_rq(cpu); /*得到要检查 cpu 所在运行队列*/ /*检查 cpu 所在运行队列的下一个要调度的实时任务优先级是否比当前队列实时任务优先级最高的任务优先级高，如果优先级更高，则表示找到了，否则进行下一次迭代找下一个，直到回到回绕到 rq->cpu 为止，此时返回 nr_cpu_ids*/ if (next_rq->rt.highest_prio.next < rq->rt.highest_prio.curr) break; } return cpu; } /*搜索下一个 cpu 总是从 rq->cpu 开始并且当我们再次搜索 rq->cpu 时结束。它绝不会返回 rq->cpu。它返回下一个要检查的 cpu，或者在循环结束时返回 nr_cpu_ids。 rq->rt.push_cpu 记录最后一个由该函数返回的 cpu，或者作为第一个开始循环的实例，此时它必须是 rq->cpu。*/ static int rto_next_cpu(struct rq *rq) { int prev_cpu = rq->rt.push_cpu; /*前次检查的 cpu 是本次搜索的起点*/ int cpu; /*从 rq 的 root domain 的实时过载 CPU 掩码中选出下一个掩码位所代表的 cpu*/ cpu = cpumask_next(prev_cpu, rq->rd->rto_mask); /*如果前一个 cpu 小于 rq 的 CPU，那么表明此前已经越过了掩码的结束边界，并且这是一轮从掩码起始开始的搜索。我们在上面得到的 cpu 大于等于 rq 的 CPU 时结束搜索。*/ if (prev_cpu < rq->cpu) { if (cpu >= rq->cpu) /*绕了一圈回到自己，结束搜索*/ return nr_cpu_ids; /*搜索结束标志为 nr_cpu_ids*/ } else if (cpu >= nr_cpu_ids) { /*我们到达掩码的边界，回绕到掩码的起始进行搜索。如果结果仍然大于或等于 rq 的 CPU，那么循环结束。*/ cpu = cpumask_first(rq->rd->rto_mask); /*回绕到掩码起始*/ if (cpu >= rq->cpu) return nr_cpu_ids; /*示意搜索结束*/ } rq->rt.push_cpu = cpu; /*记录下一个要检查推操作的 cpu*/ /* Return cpu to let the caller know if the loop is finished or not */ return cpu; } 6.2 IPI实现部分 bool irq_work_queue_on(struct irq_work *work, int cpu) { /* All work should have been flushed before going offline */ WARN_ON_ONCE(cpu_is_offline(cpu)); /* Arch remote IPI send/receive backend aren't NMI safe */ WARN_ON_ONCE(in_nmi()); /* Only queue if not already pending */ if (!irq_work_claim(work)) return false; /*将工作放入指定的 cpu 的 raised_list 链表，通过体系结构相关的函数发送 ipi 到指定 cpu*/ if (llist_add(&work->llnode, &per_cpu(raised_list, cpu))) arch_send_call_function_single_ipi(cpu); return true; } EXPORT_SYMBOL_GPL(irq_work_queue_on);

rq->rt.push_work是在初始化实时任务运行队列的时候进行的初始化:

void init_rt_rq(struct rt_rq *rt_rq) { ... #ifdef HAVE_RT_PUSH_IPI rt_rq->push_flags = 0; rt_rq->push_cpu = nr_cpu_ids; /*初始化 push_cpu*/ raw_spin_lock_init(&rt_rq->push_lock); init_irq_work(&rt_rq->push_work, push_irq_work_func); /*在此初始化 irq work*/ #endif ... }

当指定的推 CPU 收到触发的 IPI 后，会调用raised_list链表上的工作的回调函数处理 irq 工作,push_irq_work_func()在执行推任务的 CPU 收到 IPI 后异步地调用:

update_process_times() -> irq_work_tick() -> irq_work_run_list() -> push_irq_work_func() static void push_irq_work_func(struct irq_work *work) { /*这个 work 指向拉任务队列的 push_work，因此 rt_rq 是拉任务队列，也是我们要推的目标*/ struct rt_rq *rt_rq = container_of(work, struct rt_rq, push_work); /*推任务 CPU 把任务往拉任务队列 rt_rq 上推*/ try_to_push_tasks(rt_rq); } /* Called from hardirq context */ static void try_to_push_tasks(void *arg) { struct rt_rq *rt_rq = arg; /*推的目标，即拉操作的那个实时运行队列*/ struct rq *rq, *src_rq; int this_cpu; int cpu; this_cpu = rt_rq->push_cpu; /*发起推操作的 cpu 是被拉操作选中的*/ /* Paranoid check */ BUG_ON(this_cpu != smp_processor_id()); /*如果当前 cpu 不是选中 cpu 肯定是 bug*/ rq = cpu_rq(this_cpu); /*推操作所在运行队列*/ src_rq = rq_of_rt_rq(rt_rq); /*拉操作所在运行队列设为源队列*/ again: if (has_pushable_tasks(rq)) { /*如果推操作上有任务可推*/ raw_spin_lock(&rq->lock); push_rt_task(rq); /*从推队列上推一个任务出去*/ raw_spin_unlock(&rq->lock); } /*下面是把 IPI 发给下一个实时过载队列的过程*/ raw_spin_lock(&rt_rq->push_lock); /*如果此时发现源队列又有拉操作发生（比如说，优先级又降低了），需要清除 RT_PUSH_IPI_RESTART标志位，并重新开始 RT_PUSH_IPI*/ if (rt_rq->push_flags & RT_PUSH_IPI_RESTART) { rt_rq->push_flags &= ~RT_PUSH_IPI_RESTART; rt_rq->push_cpu = src_rq->cpu; /*重新开始的起点是源队列 cpu*/ } /*源队列上接着往下找推操作 cpu*/ cpu = find_next_push_cpu(src_rq); /*如果找不到了，清除 RT_PUSH_IPI_EXECUTING 标志位，准备结束 RT_PUSH_IPI*/ if (cpu >= nr_cpu_ids) rt_rq->push_flags &= ~RT_PUSH_IPI_EXECUTING; raw_spin_unlock(&rt_rq->push_lock); /*找不到推任务的 cpu 了，结束 RT_PUSH_IPI*/ if (cpu >= nr_cpu_ids) return; /*如果发生了拉操作 restart，又选中该 cpu 执行推操作，这种情况无需再次触发 IPI，只需看当前队列还有没有任务要推就可以了。*/ if (unlikely(cpu == rq->cpu)) goto again; /*否则，触发 IPI 给下一个推操作的 cpu，把拉操作的 push_work 又排入 irq 工作队列*/ irq_work_queue_on(&rt_rq->push_work, cpu); } 6.3 RT_PUSH_IPI总结只把 IPI 发给第一个过载的 CPU，当它把所有能推出去的任务推走后，再找下一个能把任务推给源 CPU 的过载的 CPU；如果 IPI 请求发给所有 RT overload list 上的 CPU，我们还会遇到相同锁竞争的问题，只不过方向相反； try_to_push_tasks()用的是push_rt_task()推走一个task，然后就把 IPI 交给下一个 cpu 了，并没有把所有能推的任务都推走，也没有指定推到哪一个cpu，而是找找到优先级最低的队列；当所有过载 CPU 搜索一遍的时候，IPI 停止；如果在这期间源 CPU 再次降低它的优先级，那么设置一个标志位，告诉 IPI 遍历需从源 CPU 之后的第一个 RT 过载 CPU 重新开始，这里的顺序是按 root domain 中 rto_mask 的位置为顺序，和 CPU 优先级没关系。 7、参考文档

1.实时调度负载均衡

2.hellokitty2 RT负载均衡

【本文地址】

公司简介

联系我们