在如今數字化浪潮洶涌的時代，無論是工業自動化生產線的精準操控，還是多媒體應用中流暢的音視頻播放體驗，又或是醫療設備對生命體征的實時監測，背后都離不開一個關鍵因素 —— 系統的實時響應能力。而 Linux 系統，作為開源世界的中流砥柱，在眾多領域廣泛應用。但默認情況下，Linux 的進程調度主要側重于公平性，對于那些對時間極為敏感的實時任務而言，原生的調度機制顯得有些力不從心。

那么，如何才能挖掘 Linux 系統的潛力，讓它在實時場景中大放異彩，顯著優化應用性能呢？今天，就帶大家走進 Linux 實時調度的實戰世界，一同探索其中的奧秘，掌握讓應用性能飛躍的實用技巧 。 

一、Linux調度概述

在日常生活中，我們常常會遇到排隊的場景。比如在超市結賬時，顧客們會在收銀臺前依次排隊等待結賬。在這個過程中，收銀員會按照排隊的順序為顧客服務，這就是一種簡單的調度方式。而在 Linux 操作系統中，也存在著類似的調度機制，它負責管理系統中的進程，決定哪個進程可以獲得 CPU 資源并運行。

進程是計算機中的程序關于某數據集合上的一次運行活動，是系統進行資源分配和調度的基本單位。在單處理器系統中，同一時刻只有一個進程能夠在 CPU 上運行，就像只有一個收銀員為顧客服務一樣。而在多處理器系統中，雖然可以有多個進程同時在不同的 CPU 上運行，但總體上仍然需要對進程進行合理的調度，以確保系統的高效運行。

Linux 系統中的進程可以分為不同的類型，其中實時進程對時間要求較高，它們需要在規定的時間內完成任務。實時進程又可以進一步分為硬實時進程和軟實時進程。硬實時進程必須在絕對的時間窗口內完成任務，否則可能會導致系統失效或災難性后果，比如航空航天控制、醫療設備等領域的任務。軟實時進程雖然也追求在規定時間內完成任務，但偶爾的超時通常不會導致系統完全失效，只會影響系統的服務質量或用戶體驗，像多媒體處理、網絡通信等場景中的任務。除了實時進程，還有普通進程，它們對時間的要求相對較低，在系統資源分配中處于相對次要的地位。

為了實現對進程的有效調度，Linux 系統采用了多種調度算法。其中，時間片輪轉調度算法是一種常見的調度方式。它將 CPU 的時間劃分為一個個固定長度的時間片，每個進程輪流獲得一個時間片來運行。當一個進程的時間片用完后，即使它還沒有完成任務，也會被暫停，然后被放入就緒隊列的末尾，等待下一輪調度。這種調度方式就像是超市里的顧客們輪流在收銀臺結賬，每個人都有機會得到服務，從而保證了系統的公平性和響應性。

實時調度器主要為了解決以下四種情況：

• 在喚醒任務時，待喚醒的任務放置到哪個運行隊列最合適（這里稱為pre-balance）；
• 新喚醒任務的優先級比某個運行隊列的當前任務更低時，怎么處理這個更低優先級任務；
• 新喚醒任務的優先級比同一運行隊列的某個任務更高時，并且搶占了該低優先級任務，該低優先級任務怎么處理？
• 當某個任務降低自身優先級，導致原來更低優先級任務相比之下具有更高優先級，這種情況怎么處理。
  對于情況2和情況3，實時調度器采用push操作。push操作從根域中所有運行隊列中挑選一個運行隊列（一個cpu對應一個運行隊列），該運行隊列的優先級比待push任務的優先級更低。運行隊列的優先級是指該運行隊列上所有任務的最高優先級。

對于情況4，實時調度器采用pull操作。當某個運行隊列上準備調度時，候選任務比當前任務的優先級更低時，實時調度器檢查其他運行隊列，確定是否可以pull更高優先級任務到本運行隊列。還有，當某個運行隊列上發生調度時，該運行隊列上沒有任務比當前任務優先級高，實時調度器執行pull操作，從其他運行隊列中pull更高優先級任務到本運行隊列。

每CPU變量運行隊列rq，包含一個rt_rq數據結構。rt_rq結構體主要內容如下：

struct rt_rq {
    struct rt_prio_array  active;
    ...
    unsigned long         rt_nr_running;         // 可運行實時任務個數
    unsigned long         rt_nr_migratory;       // 該運行隊列上可以遷移到其他運行隊列的實時任務個數
    unsigned long         rt_nr_uninterruptible;
    int                   highest_prio;
    int                   overloaded;
};

實時任務優先級范圍為0到99。這些實時任務組織成優先級索引數組active，該優先級數組的數據結構類型為rt_prio_arry。rt_prio_arry數據結構由兩部分組成，一部分是位圖，另一部分是數組。

struct rt_prio_arry {
    unsigned long bitmap[BITS_TO_LONGS(MAX_RT_PRIO+1)];
    struct list_head queue[MAX_RT_PRIO]；
}

二、實時調度策略

Linux內核中提供了兩種實時調度策略：SCHED_FIFO和SCHED_RR，其中RR是帶有時間片的FIFO。這兩種調度算法實現的都是靜態優先級。內核不為實時進程計算動態優先級。這能保證給定優先級別的實時進程總能搶占優先級比他低得進程。linux的實時調度算法提供了一種軟實時工作方式。

實時優先級范圍從0到MAX_RT_PRIO減一。默認情況下，MAX_RT_PRIO為100（定義在include/linux/sched.h中），所以默認的實時優先級范圍是從0到99。SCHED_NORMAL級進程的nice值共享了這個取值空間，它的取值范圍是從MAX_RT_PRIO到MAX_RT_PRIO+40。也就是說，在默認情況下，nice值從-20到19直接對應的是從100到139的優先級范圍，這就是普通進程的靜態優先級范圍。在實時調度策略下。schedule()函數的運行會關聯到實時調度類rt_sched_class。

2.1SCHED_FIFO：獨占 CPU 的 “霸王龍”

在 Linux 的實時調度策略中，SCHED_FIFO 就像是恐龍時代的霸王龍，霸氣十足。它采用先進先出（FIFO）的調度方式，這種方式非常直接，沒有復雜的時間片輪轉機制。一旦一個進程被調度，只要沒有更高優先級的進程出現，它就會一直霸占著 CPU 運行下去，運行時長不受任何限制。這就好比在一場比賽中，只要沒有更強的對手出現，當前的冠軍就可以一直保持領先，不會被替換下場。

以音頻處理場景為例，在實時音頻錄制和播放中，就經常會用到 SCHED_FIFO 策略。在錄制音頻時，需要保證音頻數據的連續性和及時性，不能有絲毫的延遲或中斷。如果采用 SCHED_FIFO 策略，音頻錄制進程一旦獲得 CPU 資源，就會持續運行，將麥克風采集到的音頻數據及時地寫入存儲設備。在播放音頻時，音頻播放進程也會獨占 CPU，按照順序將音頻數據從存儲設備中讀取出來，并發送到音頻輸出設備進行播放。這樣可以確保音頻的流暢播放，不會出現卡頓或雜音的情況，為用戶帶來高品質的音頻體驗。

SCHED_FIFO 策略的優點顯而易見，它可以為那些對時間要求極為嚴格的實時進程提供穩定且可預測的執行時間，這對于一些需要精確控制時間的系統來說至關重要，比如工業控制系統、機器人控制等領域。在這些系統中，任務的執行時間必須是可預測的，否則可能會導致嚴重的后果。

然而，SCHED_FIFO 策略也存在明顯的缺點。由于它沒有時間片的概念，一旦一個低優先級的進程先獲得了 CPU 資源，并且一直不主動放棄，那么其他優先級較低的進程就可能會一直處于饑餓狀態，無法獲得 CPU 資源來執行。這就好比一群人在排隊等待服務，但是排在前面的人一直占用著服務資源不離開，后面的人就只能一直等待，這顯然是不公平的。

2.2SCHED_RR：公平輪替的 “時間掌控者”

與 SCHED_FIFO 不同，SCHED_RR 像是一位公平的時間掌控者，采用時間片輪轉的調度機制。在這種策略下，每個進程都會被分配一個固定的時間片。當進程運行時，時間片會逐漸減少。一旦進程用完了自己的時間片，它就會被放入就緒隊列的末尾，同時釋放 CPU 資源，讓其他相同優先級的進程有機會執行。這就像一場接力比賽，每個選手都有固定的跑步時間，時間一到就把接力棒交給下一位選手，保證了每個選手都有公平的參與機會。

以動畫渲染場景為例，在制作動畫時，通常會有多個任務同時進行，比如模型渲染、材質處理、光影計算等。這些任務可能具有相同的優先級，需要合理地分配 CPU 資源。如果采用 SCHED_RR 策略，每個渲染任務都會被分配一個時間片。在自己的時間片內，任務可以充分利用 CPU 資源進行計算和處理。當時間片用完后，任務會暫停，將 CPU 資源讓給其他任務。這樣可以確保每個渲染任務都能得到及時的處理，不會因為某個任務長時間占用 CPU 而導致其他任務延遲，從而保證了動畫渲染的高效進行。

SCHED_RR 策略在保證實時性的同時，還兼顧了公平性。它通過時間片的輪轉，讓每個進程都能在一定的時間內獲得 CPU 資源，避免了低優先級進程長時間得不到執行的情況。這使得它在一些對響應時間要求較高，同時又需要保證公平性的實時進程中得到了廣泛應用，比如交互式應用程序、游戲等。在這些應用中，用戶希望能夠得到及時的響應，同時也不希望某個任務獨占 CPU 資源，導致其他操作變得遲緩。

三、實時調度類的數據結構詳解

3.1優先級隊列rt_prio_array

在kernel/sched.c中，是一組鏈表，每個優先級對應一個鏈表。還維護一個由101 bit組成的bitmap，其中實時進程優先級為0－99，占100 bit，再加1 bit的定界符。當某個優先級別上有進程被插入列表時，相應的比特位就被置位。 通常用sched_find_first_bit()函數查詢該bitmap，它返回當前被置位的最高優先級的數組下標。由于使用位圖，查找一個任務來執行所需要的時間并不依賴于活動任務的個數，而是依賴于優先級的數量。可見實時調度是一個O(1)調度策略。

struct rt_prio_array {
 DECLARE_BITMAP(bitmap, MAX_RT_PRIO+1); /* 包含1 bit的定界符 */
 struct list_head queue[MAX_RT_PRIO];
};

這里用include/linux/types.h中的DECLARE_BITMAP宏來定義指定長度的位圖，用include/linux/list.h中的struct list_head來為100個優先級定義各自的雙鏈表。在實時調度中，運行進程根據優先級放到對應的隊列里面，對于相同的優先級的進程后面來的進程放到同一優先級隊列的隊尾。對于FIFO/RR調度，各自的進程需要設置相關的屬性。進程運行時，要根據task中的這些屬性判斷和設置，放棄cpu的時機(運行完或是時間片用完)。

3.2實時運行隊列rt_rq

在kernel/sched.c中，用于組織實時調度的相關信息。

struct rt_rq {
 struct rt_prio_array active;
 unsigned long rt_nr_running;
#if defined CONFIG_SMP || defined CONFIG_RT_GROUP_SCHED
 struct {
 int curr; /* 最高優先級的實時任務 */
#ifdef CONFIG_SMP
 int next; /* 下一個最高優先級的任務 */
#endif
 } highest_prio;
#endif
#ifdef CONFIG_SMP
 unsigned long rt_nr_migratory;
 unsigned long rt_nr_total;
 int overloaded;
 struct plist_head pushable_tasks;
#endif
 int rt_throttled;
 u64 rt_time;
 u64 rt_runtime;
 /* Nests inside the rq lock: */
 spinlock_t rt_runtime_lock;

#ifdef CONFIG_RT_GROUP_SCHED
 unsigned long rt_nr_boosted;

 struct rq *rq;
 struct list_head leaf_rt_rq_list;
 struct task_group *tg;
 struct sched_rt_entity *rt_se;
#endif
};

3.3實時調度實體 sched_rt_entity

在 Linux 內核的實時調度機制中，sched_rt_entity結構體扮演著至關重要的角色，它就像是一個精心打造的 “任務名片”，記錄了實時進程參與調度所需的關鍵信息。該結構體定義于include/linux/sched.h頭文件中，其源碼如下：

struct sched_rt_entity {
    struct list_head run_list;         // 用于將“實時調度實體”加入到優先級隊列中的
    unsigned long timeout;            // 用于設置調度超時時間
    unsigned long watchdog_stamp;     // 用于記錄jiffies的值
    unsigned int time_slice;         // 時間片
    unsigned short on_rq;
    unsigned short on_list;
    struct sched_rt_entity *back;     // 用于由上到下連接“實時調度實體”
#ifdef CONFIG_RT_GROUP_SCHED
    struct sched_rt_entity *parent;   // 指向父類“實時調度實體”
    /* rq on which this entity is (to be) queued: */
    struct rt_rq *rt_rq;             // 表示“實時調度實體”所屬的“實時運行隊列”
    /* rq "owned" by this entity/group: */
    struct rt_rq *my_q;              // 表示“實時調度實體”所擁有的“實時運行隊列”，用于管理“子任務”
#endif
} __randomize_layout;

run_list字段是一個雙向鏈表節點，它就像一根無形的線，將各個實時調度實體按照優先級串聯起來，加入到優先級隊列中，方便調度器快速定位和處理。當一個實時進程被創建或者狀態發生變化時，它的run_list就會被插入到相應的優先級隊列中，等待調度器的調度。

timeout字段用于設置調度超時時間，這就像是給任務設定了一個 “鬧鐘”。當任務運行時間超過這個設定的超時時間時，調度器可能會對其進行特殊處理，比如將其從 CPU 上移除，重新調度其他任務，以確保系統的實時性和穩定性。在一些對時間要求極高的實時系統中，如自動駕駛汽車的控制系統，每個任務都必須在規定的時間內完成，否則可能會導致嚴重的后果。timeout字段就可以保證這些任務不會因為長時間占用 CPU 而影響其他關鍵任務的執行。

watchdog_stamp字段用于記錄jiffies的值，jiffies是 Linux 內核中的一個全局變量，表示系統啟動以來的時鐘滴答數。通過記錄jiffies的值，watchdog_stamp可以為調度器提供時間參考，用于判斷任務的運行狀態和調度時機。比如，調度器可以根據watchdog_stamp和當前的jiffies值來計算任務的運行時間，從而決定是否需要對任務進行調度。

time_slice字段表示時間片，對于采用時間片輪轉調度策略（如SCHED_RR）的實時進程來說，這個字段尤為重要。它規定了每個進程在被調度后可以連續運行的時間長度。當進程的時間片用完后，調度器會將其從 CPU 上移除，并將其放入就緒隊列的末尾，等待下一輪調度。這就像一場接力比賽，每個選手都有固定的跑步時間，時間一到就把接力棒交給下一位選手，保證了每個選手都有公平的參與機會。在多媒體播放系統中，音頻和視頻的解碼任務通常采用SCHED_RR策略，通過合理設置time_slice，可以確保音頻和視頻的流暢播放，不會出現卡頓或延遲的情況。

在支持實時組調度（CONFIG_RT_GROUP_SCHED）的情況下，parent字段指向父類 “實時調度實體”，這就像是一個家族樹中的父子關系，通過這種關系，調度器可以更好地管理和調度整個任務組。rt_rq字段表示 “實時調度實體” 所屬的 “實時運行隊列”，而my_q字段則表示 “實時調度實體” 所擁有的 “實時運行隊列”，用于管理 “子任務”。這種層次化的結構設計，使得調度器能夠更加靈活地處理復雜的實時任務場景。

3.4實時就緒隊列 struct rt_rq

struct rt_rq結構體是 Linux 內核實時調度的核心數據結構之一，它就像是一個高效的 “任務指揮官”，負責管理實時進程的運行隊列，在核心調度器管理活動進程中發揮著舉足輕重的作用。該結構體定義于kernel/sched/sched.h頭文件中，其源碼如下：

struct rt_rq {
    struct rt_prio_array active;        // 優先級隊列
    unsigned int rt_nr_running;        // 在RT運行隊列中所有活動的任務數
    unsigned int rr_nr_running;
#if defined CONFIG_SMP || defined CONFIG_RT_GROUP_SCHED
    struct {
        int curr;                       // 當前RT任務的最高優先級
#ifdef CONFIG_SMP
        int next;                       // 下一個要運行的RT任務的優先級，如果兩個任務都有最高優先級，則curr == next
#endif
    } highest_prio;
#endif
#ifdef CONFIG_SMP
    unsigned long rt_nr_migratory;     // 任務沒有綁定在某個CPU上時，這個值會增減，用于任務遷移
    unsigned long rt_nr_total;         // 用于overload檢查
    int overloaded;                    // RT運行隊列過載，則將任務推送到其他CPU
    struct plist_head pushable_tasks;  // 優先級列表，用于推送過載任務
#endif /* CONFIG_SMP */
    int rt_queued;                     // 表示RT運行隊列已經加入rq隊列
    int rt_throttled;                  // 用于限流操作
    u64 rt_time;                       // 累加的運行時，超出了本地rt_runtime時，則進行限制
    u64 rt_runtime;                    // 分配給本地池的運行時
    /* Nests inside the rq lock: */
    raw_spinlock_t rt_runtime_lock;
#ifdef CONFIG_RT_GROUP_SCHED
    unsigned long rt_nr_boosted;       // 用于優先級翻轉問題解決
    struct rq *rq;                     // 指向運行隊列
    struct task_group *tg;             // 指向任務組
#endif
};

active字段是一個rt_prio_array類型的優先級隊列，它維護了 100 個優先級的隊列（鏈表），優先級范圍從 0 到 99，從高到低排列。同時，它還定義了位圖，用于快速查詢。這就像是一個多層的貨架，每個貨架層對應一個優先級，實時進程根據其優先級被放置在相應的貨架層上。調度器可以通過位圖快速找到最高優先級的隊列，從而選擇優先級最高的進程進行調度，大大提高了調度效率。在航空航天控制系統中，各種實時任務的優先級劃分非常嚴格，通過active優先級隊列，調度器能夠快速響應高優先級任務，確保系統的安全和穩定運行。

rt_nr_running字段表示在 “實時運行隊列” 中所有活動的任務數，這個數字就像是一個實時監控的計數器，調度器可以根據它來了解當前實時運行隊列中的任務負載情況。如果任務數過多，調度器可能會采取一些措施，如任務遷移、限流等，以保證系統的正常運行。

在支持對稱多處理（CONFIG_SMP）或實時組調度（CONFIG_RT_GROUP_SCHED）的情況下，highest_prio結構體中的curr字段表示當前 RT 任務的最高優先級，next字段表示下一個要運行的 RT 任務的優先級。如果兩個任務都有最高優先級，則curr和next字段值相等。這些字段就像是調度器的 “指南針”，幫助調度器在眾多任務中準確地選擇下一個要運行的任務。

rt_nr_migratory字段用于記錄任務沒有綁定在某個 CPU 上時，這個值會增減，用于任務遷移。在多處理器系統中，當某個 CPU 的負載過高時，調度器可以根據這個字段的值，將一些可遷移的任務遷移到其他 CPU 上，以實現負載均衡。rt_nr_total字段用于overload檢查，當rt_nr_total超過一定閾值時，說明系統可能處于過載狀態，調度器會采取相應的措施，如將任務推送到其他 CPU，以緩解系統壓力。overloaded字段表示 RT 運行隊列過載，當該字段為真時，調度器會將任務推送到其他 CPU，以保證系統的正常運行。pushable_tasks字段是一個優先級列表，用于推送過載任務，它就像是一個 “任務搬運工”，將過載的任務從一個 CPU 推送到其他 CPU 上。

rt_queued字段表示 RT 運行隊列已經加入rq隊列，rq隊列是系統中所有進程的運行隊列，RT 運行隊列是其中的一部分。rt_throttled字段用于限流操作，當實時進程的運行時間超過一定限制時，調度器會對其進行限流，以保證系統的公平性和穩定性。rt_time字段表示累加的運行時，當超出本地rt_runtime時，則進行限制。rt_runtime字段表示分配給本地池的運行時，它就像是一個 “資源配額”，限制了實時進程在本地的運行時間。

在支持實時組調度（CONFIG_RT_GROUP_SCHED）的情況下，rt_nr_boosted字段用于優先級翻轉問題解決。在實時系統中，可能會出現優先級翻轉的情況，即低優先級任務持有高優先級任務所需的資源，導致高優先級任務無法執行。通過rt_nr_boosted字段，調度器可以對任務的優先級進行調整，解決優先級翻轉問題。rq字段指向運行隊列，tg字段指向任務組，通過這些指針，調度器可以更好地管理和調度整個任務組。

實時調度的主要操作：實時調度的操作在kernel/sched_rt.c中實現。

（1）進程插入enqueue_task_rt：更新調度信息，調用enqueue_rt_entity()-->__enqueue_rt_entity()，將調度實體插入到相應優先級隊列的末尾。如下：

static void
enqueue_task_rt(struct rq *rq, struct task_struct *p, int wakeup, bool head)
{
 struct sched_rt_entity *rt_se = &p->rt;

 if (wakeup)
 rt_se->timeout = 0;

 enqueue_rt_entity(rt_se, head); /* 實際工作 */

 if (!task_current(rq, p) && p->rt.nr_cpus_allowed > 1)
 enqueue_pushable_task(rq, p); /* 添加到對應的hash表中 */
}

static void enqueue_rt_entity(struct sched_rt_entity *rt_se, bool head)
{
 dequeue_rt_stack(rt_se); /* 先從運行隊列中刪除 */
 for_each_sched_rt_entity(rt_se)
 __enqueue_rt_entity(rt_se, head); /* 然后添加到運行隊列尾部 */
}

static void __enqueue_rt_entity(struct sched_rt_entity *rt_se, bool head)
{
 struct rt_rq *rt_rq = rt_rq_of_se(rt_se);
 struct rt_prio_array *array = &rt_rq->active;
 struct rt_rq *group_rq = group_rt_rq(rt_se);
 struct list_head *queue = array->queue + rt_se_prio(rt_se);

 /*
 * Don't enqueue the group if its throttled, or when empty.
 * The latter is a consequence of the former when a child group
 * get throttled and the current group doesn't have any other
 * active members.
 */
 if (group_rq && (rt_rq_throttled(group_rq) || !group_rq->rt_nr_running))
 return;

 if (head)
 list_add(&rt_se->run_list, queue);
 else
 list_add_tail(&rt_se->run_list, queue);
 __set_bit(rt_se_prio(rt_se), array->bitmap);

 inc_rt_tasks(rt_se, rt_rq); /* 運行進程數增一 */
}

該函數先獲取運行隊列中的優先級隊列，然后調用include/linux/list.h:list_add_tail()--->__list_add()，將進程插入到鏈表的末尾。如下：

static inline void __list_add(struct list_head *new,
      struct list_head *prev,
      struct list_head *next)
{
 next->prev = new;
 new->next = next;
 new->prev = prev;
 prev->next = new;
}

（2）進程選擇pick_next_task_rt：實時調度會選擇最高優先級的實時進程來運行。調用_pick_next_task_rt()--->pick_next_rt_entity()來完成獲取下一個進程的工作。如下：

static struct task_struct *pick_next_task_rt(struct rq *rq)
{
 struct task_struct *p = _pick_next_task_rt(rq); /* 實際工作 */

 /* The running task is never eligible for pushing */
 if (p)
 dequeue_pushable_task(rq, p);

#ifdef CONFIG_SMP
 /*
 * We detect this state here so that we can avoid taking the RQ
 * lock again later if there is no need to push
 */
 rq->post_schedule = has_pushable_tasks(rq);
#endif

 return p;
}

static struct task_struct *_pick_next_task_rt(struct rq *rq)
{
 struct sched_rt_entity *rt_se;
 struct task_struct *p;
 struct rt_rq *rt_rq;

 rt_rq = &rq->rt;

 if (unlikely(!rt_rq->rt_nr_running))
 return NULL;

 if (rt_rq_throttled(rt_rq))
 return NULL;

 do { /* 遍歷組調度中的每個進程 */
 rt_se = pick_next_rt_entity(rq, rt_rq);
 BUG_ON(!rt_se);
 rt_rq = group_rt_rq(rt_se);
 } while (rt_rq);

 p = rt_task_of(rt_se);
 /* 更新執行域 */
 p->se.exec_start = rq->clock_task;

 return p;
}

static struct sched_rt_entity *pick_next_rt_entity(struct rq *rq,
   struct rt_rq *rt_rq)
{
 struct rt_prio_array *array = &rt_rq->active;
 struct sched_rt_entity *next = NULL;
 struct list_head *queue;
 int idx;
 /* 找到第一個可用的 */
 idx = sched_find_first_bit(array->bitmap);
 BUG_ON(idx >= MAX_RT_PRIO);
 /* 從鏈表組中找到對應的鏈表 */
 queue = array->queue + idx;
 next = list_entry(queue->next, struct sched_rt_entity, run_list);
 /* 返回找到的運行實體 */
 return next;
}

該函數調用include/asm-generic/bitops/sched.h:sched_find_first_bit()返回位圖中當前被置位的最高優先級，以作為這組鏈表的數組下標找到其優先級隊列。然后調用include/linux/list.h:list_entry()--->include/linux/kernel.h:container_of()，返回該優先級隊列中的第一個進程，以作為下一個要運行的實時進程。例如當前所有實時進程中最高優先級為45（換句話說，系統中沒有任何實時進程的優先級小于45），則直接讀取rt_prio_array中的queue[45]，得到優先級為45的進程隊列指針。該隊列頭上的第一個進程就是被選中的進程。這種算法的復雜度為O(1)。

sched_find_first_bit的實現如下。它與CPU體系結構相關，其他體系結構會實現自己的sched_find_fist_bit函數。下面的實現以最快的方式搜索100 bit的位圖，它能保證100 bit中至少有一位被清除。

static inline int sched_find_first_bit(const unsigned long *b)
{
#if BITS_PER_LONG == 64
 if (b[0])
 return __ffs(b[0]);
 return __ffs(b[1]) + 64;
#elif BITS_PER_LONG == 32
 if (b[0])
 return __ffs(b[0]);
 if (b[1])
 return __ffs(b[1]) + 32;
 if (b[2])
 return __ffs(b[2]) + 64;
 return __ffs(b[3]) + 96;
#else
#error BITS_PER_LONG not defined
#endif
}

（3）進程刪除dequeue_task_rt：從優先級隊列中刪除實時進程，并更新調度信息，然后把這個進程添加到隊尾。調用鏈為dequeue_rt_entity()--->dequeue_rt_stack()--->__dequeue_rt_entity()，如下：

static void dequeue_task_rt(struct rq *rq, struct task_struct *p, int sleep)
{
 struct sched_rt_entity *rt_se = &p->rt;
 /* 更新調度信息 */
 update_curr_rt(rq);
 /* 實際工作，將rt_se從運行隊列中刪除然后 
      添加到隊列尾部 */
 dequeue_rt_entity(rt_se);
 /* 從hash表中刪除 */
 dequeue_pushable_task(rq, p);
}

static void update_curr_rt(struct rq *rq)
{
 struct task_struct *curr = rq->curr;
 struct sched_rt_entity *rt_se = &curr->rt;
 struct rt_rq *rt_rq = rt_rq_of_se(rt_se);
 u64 delta_exec;

 if (!task_has_rt_policy(curr)) /* 判斷是否問實時調度進程 */
 return;
 /* 執行時間 */
 delta_exec = rq->clock_task - curr->se.exec_start;
 if (unlikely((s64)delta_exec < 0))
 delta_exec = 0;

 schedstat_set(curr->se.exec_max, max(curr->se.exec_max, delta_exec));
 /* 更新當前進程的總的執行時間 */
 curr->se.sum_exec_runtime += delta_exec;
 account_group_exec_runtime(curr, delta_exec);
 /* 更新執行的開始時間 */
 curr->se.exec_start = rq->clock_task;
 cpuacct_charge(curr, delta_exec); /* 組調度相關 */

 sched_rt_avg_update(rq, delta_exec);

 if (!rt_bandwidth_enabled())
 return;

 for_each_sched_rt_entity(rt_se) {
 rt_rq = rt_rq_of_se(rt_se);

 if (sched_rt_runtime(rt_rq) != RUNTIME_INF) {
 spin_lock(&rt_rq->rt_runtime_lock);
 rt_rq->rt_time += delta_exec;
 if (sched_rt_runtime_exceeded(rt_rq))
 resched_task(curr);
 spin_unlock(&rt_rq->rt_runtime_lock);
 }
 }
}

static void dequeue_rt_entity(struct sched_rt_entity *rt_se)
{
 dequeue_rt_stack(rt_se); /* 從運行隊列中刪除 */

 for_each_sched_rt_entity(rt_se) {
 struct rt_rq *rt_rq = group_rt_rq(rt_se);

 if (rt_rq && rt_rq->rt_nr_running)
 __enqueue_rt_entity(rt_se, false); /* 添加到隊尾 */
 }
}

static void dequeue_rt_stack(struct sched_rt_entity *rt_se)
{
 struct sched_rt_entity *back = NULL;

 for_each_sched_rt_entity(rt_se) { /* 遍歷整個組調度實體 */
 rt_se->back = back; /* 可見rt_se的back實體為組調度中前一個調度實體 */
 back = rt_se;
 }
 /* 將組中的所有進程從運行隊列中移除 */
 for (rt_se = back; rt_se; rt_se = rt_se->back) {
 if (on_rt_rq(rt_se))
 __dequeue_rt_entity(rt_se);
 }
}

static void __dequeue_rt_entity(struct sched_rt_entity *rt_se)
{
 struct rt_rq *rt_rq = rt_rq_of_se(rt_se);
 struct rt_prio_array *array = &rt_rq->active;
 /* 移除進程 */
 list_del_init(&rt_se->run_list);
 /* 如果鏈表變為空，則將位圖中對應的bit位清零 */
 if (list_empty(array->queue + rt_se_prio(rt_se)))
 __clear_bit(rt_se_prio(rt_se), array->bitmap);

 dec_rt_tasks(rt_se, rt_rq); /* 運行進程計數減一 */
}

可見更新調度信息的函數為update_curr_rt()，在dequeue_rt_entity()中將當前實時進程從運行隊列中移除，并添加到隊尾。完成工作函數為dequeue_rt_stack()--->__dequeue_rt_entity()，它調用include/linux/list.h:list_del_init()--->__list_del()刪除進程。然后如果鏈表變為空，則將位圖中對應優先級的bit位清零。如下：

static inline void __list_del(struct list_head * prev, struct list_head * next)
{
 next->prev = prev;
 prev->next = next;
}

從上面的介紹可以看出，對于實時調度，Linux的實現比較簡單，仍然采用之前的O(1)調度策略，把所有的運行進程根據優先級放到不用的隊列里面，采用位圖方式進行使用記錄。進隊列僅僅是刪除原來隊列里面的本進程，然后將他掛到隊列尾部；而對于“移除”操作，也僅僅是從隊列里面移除后添加到運行隊列尾部。

四、實時調度類在實際中的應用

4.1工業控制系統：精準控制的幕后英雄

在工業 4.0 的浪潮下，工業控制系統正朝著智能化、自動化的方向飛速發展。從汽車制造到電子設備生產，自動化生產線已經成為現代工業的核心。在這些復雜的生產線上，各種設備協同工作，每一個動作、每一次數據傳輸都需要精確的時間控制。Linux 實時調度類在其中扮演著至關重要的角色，它就像是一位精準的指揮官，確保每一個任務都能按時執行，從而實現整個生產線的高效、穩定運行。

以汽車制造為例，自動化生產線涉及眾多復雜的工序，如車身焊接、零部件裝配、噴漆等。在車身焊接環節，機械臂需要按照精確的時間順序和位置坐標進行焊接操作。如果焊接任務不能按時完成，可能會導致車身結構不穩定，影響汽車的質量和安全性。

Linux 實時調度類通過采用 SCHED_FIFO 或 SCHED_RR 策略，為焊接任務分配高優先級，確保機械臂能夠及時響應控制指令，準確地完成焊接操作。同時，在生產線的物料運輸環節，AGV（自動導引車）需要根據生產進度及時將零部件運輸到指定位置。Linux 實時調度類可以根據 AGV 的任務優先級和實時路況，合理地調度 AGV 的運行，避免出現交通擁堵和任務延遲的情況，保證生產線的物料供應順暢。

據相關數據顯示，采用 Linux 實時調度類的工業控制系統，生產效率能夠提升 20% 以上，產品次品率降低 15% 左右。這充分證明了 Linux 實時調度類在工業控制系統中的重要性和價值，它為工業生產的精準控制提供了堅實的技術保障，是工業自動化不可或缺的關鍵技術之一。

4.2多媒體處理：流暢體驗的保障

在當今的數字時代，多媒體應用已經深入到我們生活的方方面面，從高清視頻播放到音頻實時處理，從視頻會議到游戲娛樂，我們對多媒體體驗的要求越來越高。而 Linux 實時調度類正是實現流暢多媒體體驗的幕后功臣，它對時間敏感性任務的精確調度，確保了多媒體處理的流暢性、穩定性和實時性。

以高清視頻播放為例，在播放高清視頻時，視頻解碼和音頻解碼任務需要在極短的時間內完成，以保證視頻和音頻的同步播放，避免出現卡頓和延遲的情況。Linux 實時調度類可以為視頻解碼和音頻解碼任務分配較高的優先級，并采用合適的調度策略，如 SCHED_RR 策略，為每個任務分配固定的時間片，確保它們能夠及時獲取 CPU 資源，快速地完成解碼工作。同時，在視頻渲染和音頻輸出環節，實時調度類也能保證任務的及時執行，將解碼后的視頻幀和音頻數據快速地輸出到顯示設備和音頻設備上，為用戶帶來流暢、清晰的視聽體驗。

在音頻實時處理領域，如語音識別、音頻編輯等應用中，Linux 實時調度類同樣發揮著重要作用。在語音識別過程中，麥克風采集到的語音信號需要及時進行處理和分析，以準確識別用戶的語音指令。Linux 實時調度類通過對語音處理任務的優先調度，能夠快速地對語音信號進行采樣、濾波、特征提取等操作，提高語音識別的準確率和實時性。在音頻編輯軟件中，當用戶對音頻進行剪輯、混音等操作時，實時調度類可以確保音頻處理任務的高效執行，讓用戶能夠實時聽到處理后的音頻效果，提升音頻編輯的效率和體驗。

五、配置與優化指南

5.1系統調用設置：掌控調度的 “魔法棒”

在 Linux 系統中，要實現對線程或進程調度策略和優先級的精確控制，pthread_setschedparam和sched_setscheduler這兩個系統調用就像是掌控調度的 “魔法棒”，發揮著關鍵作用。

pthread_setschedparam主要用于設置線程的調度參數，其函數原型如下：

#include <pthread.h>
int pthread_setschedparam(pthread_t thread, int policy, const struct sched_param *param);

thread參數表示目標線程的標識符，它就像是線程的 “身份證”，通過這個標識符，系統能夠準確地定位到需要設置調度參數的線程。policy參數用于指定調度策略，可取值包括SCHED_OTHER（普通分時調度策略）、SCHED_FIFO（先進先出實時調度策略）和SCHED_RR（時間片輪轉實時調度策略）等，不同的取值決定了線程在系統中的調度方式。param是一個指向struct sched_param結構體的指針，該結構體中包含了線程的優先級信息，通過設置param->sched_priority的值，可以調整線程的優先級。例如：

#include <pthread.h>
#include <sched.h>
#include <stdio.h>

void* thread_function(void* arg) {
    // 獲取線程的tid
    pid_t tid = gettid();
    // 定義調度策略和優先級變量
    int policy;
    struct sched_param sched_param;
    // 獲取當前線程的調度參數
    pthread_getschedparam(pthread_self(), &policy, &sched_param);
    // 打印當前的調度策略和優先級
    printf("Current thread (tid: %d) policy: %d priority: %d\n", tid, policy, sched_param.sched_priority);

    // 設置線程的調度策略為FIFO
    policy = SCHED_FIFO;
    // 設置線程的優先級為最大值
    sched_param.sched_priority = sched_get_priority_max(policy);
    // 設置線程的調度策略和優先級
    pthread_setschedparam(pthread_self(), policy, &sched_param);

    // 再次獲取并打印調度參數
    pthread_getschedparam(pthread_self(), &policy, &sched_param);
    printf("New thread (tid: %d) policy: %d priority: %d\n", tid, policy, sched_param.sched_priority);

    // 線程工作代碼...
    return NULL;
}

int main() {
    pthread_t thread;
    // 創建線程
    pthread_create(&thread, NULL, &thread_function, NULL);
    // 等待線程結束
    pthread_join(thread, NULL);
    return 0;
}

在上述示例中，首先獲取當前線程的調度參數并打印，然后將線程的調度策略設置為SCHED_FIFO，優先級設置為該策略允許的最大值，最后再次獲取并打印調度參數，以驗證設置是否生效。

sched_setscheduler函數則用于設置進程的調度策略和優先級，其函數原型為：

#include <sched.h>
int sched_setscheduler(pid_t pid, int policy, const struct sched_param *param);

pid參數表示要設置調度策略的進程的進程 ID，如果pid為 0，則表示當前進程。policy和param的含義與pthread_setschedparam中的類似。例如，將當前進程的調度策略設置為SCHED_RR，并設置優先級為 50，可以這樣實現：

#include <sched.h>
#include <stdio.h>
#include <stdlib.h>

int main() {
    struct sched_param param;
    param.sched_priority = 50;
    int ret = sched_setscheduler(0, SCHED_RR, ?m);
    if (ret == -1) {
        perror("Failed to set scheduler");
        exit(EXIT_FAILURE);
    }
    printf("Successfully set scheduler to SCHED_RR with priority 50\n");
    return 0;
}

通過這兩個系統調用，開發者可以根據實際需求靈活地設置線程或進程的調度策略和優先級，從而優化系統的性能和實時性。在實際應用中，需要根據任務的特點和系統的要求，謹慎選擇調度策略和優先級，以確保系統的穩定運行和高效執行。

5.2性能優化建議：提升效率的 “秘籍”

在 Linux 實時調度中，為了充分發揮系統的性能，提升實時任務的執行效率，我們可以從以下幾個方面入手進行優化。

首先，根據任務特點選擇調度策略是關鍵的一步。對于那些對時間要求極為嚴格，需要在最短時間內完成的任務，如工業控制系統中的緊急控制任務、航空航天中的關鍵飛行控制指令處理等，應優先考慮使用SCHED_FIFO策略。因為該策略能夠確保任務一旦獲得 CPU 資源，就可以一直運行，直到完成或者被更高優先級的任務搶占，從而保證了任務執行的及時性和連續性。

而對于那些需要公平分配 CPU 時間，且對響應時間有一定要求的任務，如多媒體播放中的音頻和視頻同步處理、交互式應用程序中的用戶輸入響應等，SCHED_RR策略則更為合適。它通過時間片輪轉的方式，讓每個任務都能在一定的時間內獲得 CPU 資源，避免了某個任務長時間獨占 CPU，導致其他任務無法及時執行的情況，兼顧了公平性和實時性。

合理分配優先級也是優化實時調度性能的重要手段。在一個復雜的實時系統中，通常會有多個不同類型的任務同時運行，這些任務對時間的敏感度和重要性各不相同。因此，我們需要根據任務的實際需求，為它們分配合理的優先級。高優先級應分配給那些對系統正常運行至關重要，且時間要求緊迫的任務，例如在醫療設備控制系統中，生命體征監測和緊急治療控制任務就需要設置較高的優先級，以確保能夠及時響應患者的生命體征變化，保障患者的生命安全。

而對于一些相對次要的任務，如系統日志記錄、數據備份等，可以分配較低的優先級，讓它們在系統資源空閑時執行，避免影響關鍵任務的執行。同時，要注意避免出現優先級反轉的情況，即低優先級任務持有高優先級任務所需的資源，導致高優先級任務無法執行。可以通過采用優先級繼承、優先級天花板等算法來解決這個問題。

優化系統資源配置也不容忽視。一方面，要合理分配 CPU 資源。在多處理器系統中，可以根據任務的特點和 CPU 的負載情況，將任務綁定到特定的 CPU 核心上執行，以提高 CPU 緩存的命中率，減少任務在不同 CPU 核心之間切換帶來的開銷。例如，對于一些計算密集型的實時任務，可以將它們固定分配到性能較強的 CPU 核心上，以充分發揮 CPU 的計算能力。

另一方面，要關注內存資源的管理。實時任務通常對內存的訪問速度和穩定性有較高的要求，因此可以通過優化內存分配算法、減少內存碎片等方式，提高內存的使用效率和性能。此外，還可以采用內存鎖定技術，將關鍵的實時任務所需的內存頁面鎖定在物理內存中，避免它們被交換到磁盤上，從而提高任務的執行速度和實時性。在一些對響應時間要求極高的金融交易系統中，就可以采用內存鎖定技術，確保交易處理任務能夠快速、穩定地執行。