10.3. 前和后半部

10.3.?前和后半部

中斷處理的一個(gè)主要問題是如何在處理中進(jìn)行長時(shí)間的任務(wù). 常常大量的工作必須響應(yīng)一個(gè)設(shè)備中斷來完成, 但是中斷處理需要很快完成并且不使中斷阻塞太長. 這 2 個(gè)需要(工作和速度)彼此沖突, 留給驅(qū)動(dòng)編寫者一點(diǎn)困擾.

Linux (許多其他系統(tǒng)一起)解決這個(gè)問題通過將中斷處理分為 2 半. 所謂的前半部是實(shí)際響應(yīng)中斷的函數(shù) -- 你使用 request_irq 注冊的那個(gè). 后半部是由前半部調(diào)度來延后執(zhí)行的函數(shù), 在一個(gè)更安全的時(shí)間. 最大的不同在前半部處理和后半部之間是所有的中斷在后半部執(zhí)行時(shí)都使能 -- 這就是為什么它在一個(gè)更安全時(shí)間運(yùn)行. 在典型的場景中, 前半部保存設(shè)備數(shù)據(jù)到一個(gè)設(shè)備特定的緩存, 調(diào)度它的后半部, 并且退出: 這個(gè)操作非常快. 后半部接著進(jìn)行任何其他需要的工作, 例如喚醒進(jìn)程, 啟動(dòng)另一個(gè) I/O 操作, 等等. 這種設(shè)置允許前半部來服務(wù)一個(gè)新中斷而同時(shí)后半部仍然在工作.

幾乎每個(gè)認(rèn)真的中斷處理都這樣劃分. 例如, 當(dāng)一個(gè)網(wǎng)絡(luò)接口報(bào)告有新報(bào)文到達(dá), 處理者只是獲取數(shù)據(jù)并且上推給協(xié)議層; 報(bào)文的實(shí)際處理在后半部進(jìn)行.

Linux 內(nèi)核有 2 個(gè)不同的機(jī)制可用來實(shí)現(xiàn)后半部處理, 我們都在第 7 章介紹. tasklet 常常是后半部處理的首選機(jī)制; 它們非?？? 但是所有的 tasklet 代碼必須是原子的. tasklet 的可選項(xiàng)是工作隊(duì)列, 它可能有一個(gè)更高的運(yùn)行周期但是允許睡眠.

下面的討論再次使用 short 驅(qū)動(dòng). 當(dāng)使用一個(gè)模塊選項(xiàng)加載時(shí), short 能夠被告知在前/后半部模式使用一個(gè) tasklet 或者工作隊(duì)列處理者來進(jìn)行中斷處理. 在這個(gè)情況下, 前半部快速地執(zhí)行; 它簡單地記住當(dāng)前時(shí)間并且調(diào)度后半部處理. 后半部接著負(fù)責(zé)將時(shí)間編碼并且喚醒任何可能在等待數(shù)據(jù)的用戶進(jìn)程.

10.3.1.?Tasklet 實(shí)現(xiàn)

記住 tasklet 是一個(gè)特殊的函數(shù), 可能被調(diào)度來運(yùn)行, 在軟中斷上下文, 在一個(gè)系統(tǒng)決定的安全時(shí)間中. 它們可能被調(diào)度運(yùn)行多次, 但是 tasklet 調(diào)度不累積; ; tasklet 只運(yùn)行一次, 即便它在被投放前被重復(fù)請求. 沒有 tasklet 會(huì)和它自己并行運(yùn)行, 因?yàn)樗贿\(yùn)行一次, 但是 tasklet 可以與 SMP 系統(tǒng)上的其他 tasklet 并行運(yùn)行. 因此, 如果你的驅(qū)動(dòng)有多個(gè) tasklet, 它們必須采取某類加鎖來避免彼此沖突.

tasklet 也保證作為函數(shù)運(yùn)行在第一個(gè)調(diào)度它們的同一個(gè) CPU 上. 因此, 一個(gè)中斷處理可以確保一個(gè) tasklet 在處理者結(jié)束前不會(huì)開始執(zhí)行. 但是, 另一個(gè)中斷當(dāng)然可能在 tasklet 在運(yùn)行時(shí)被遞交, 因此, tasklet 和中斷處理之間加鎖可能仍然需要.

tasklet 必須使用 DECLARE_TASKLET 宏來聲明:

DECLARE_TASKLET(name, function, data);

name 是給 tasklet 的名子, function 是調(diào)用來執(zhí)行 tasklet (它帶一個(gè) unsigned long 參數(shù)并且返回 void )的函數(shù), 以及 data 是一個(gè) unsigned long 值來傳遞給 tasklet 函數(shù).

short 驅(qū)動(dòng)聲明它的 tasklet 如下:

void short_do_tasklet(unsigned long);
DECLARE_TASKLET(short_tasklet, short_do_tasklet, 0);

函數(shù) tasklet_schedule 用來調(diào)度一個(gè) tasklet 運(yùn)行. 如果 short 使用 tasklet=1 來加載, 它安裝一個(gè)不同的中斷處理來保存數(shù)據(jù)并且調(diào)度 tasklet 如下:

irqreturn_t short_tl_interrupt(int irq, void *dev_id, struct pt_regs *regs)
{
        do_gettimeofday((struct timeval *) tv_head); /* cast to stop 'volatile' warning
                         */
        short_incr_tv(&tv_head);
        tasklet_schedule(&short_tasklet);
        short_wq_count++; /* record that an interrupt arrived */
        return IRQ_HANDLED;
}

實(shí)際的 tasklet 函數(shù), short_do_tasklet, 將在系統(tǒng)方便時(shí)很快執(zhí)行. 如同前面提過, 這個(gè)函數(shù)進(jìn)行處理中斷的大量工作; 它看來如此:

void short_do_tasklet (unsigned long unused)
{
        int savecount = short_wq_count, written;
        short_wq_count = 0; /* we have already been removed from the queue */
        /*
        * The bottom half reads the tv array, filled by the top half,
        * and prints it to the circular text buffer, which is then consumed
        * by reading processes */
        /* First write the number of interrupts that occurred before this bh */
        written = sprintf((char *)short_head,"bh after %6i\n",savecount);
        short_incr_bp(&short_head, written);
        /*
        * Then, write the time values. Write exactly 16 bytes at a time,
        * so it aligns with PAGE_SIZE */

        do {
                written = sprintf((char *)short_head,"%08u.%06u\n",
                                  (int)(tv_tail->tv_sec % 100000000),
                                  (int)(tv_tail->tv_usec));
                short_incr_bp(&short_head, written);
                short_incr_tv(&tv_tail);
        } while (tv_tail != tv_head);

        wake_up_interruptible(&short_queue); /* awake any reading process */
}

在別的東西中, 這個(gè) tasklet 記錄了從它上次被調(diào)用以來有多少中斷到達(dá). 一個(gè)如 short 一樣的設(shè)備能夠在短時(shí)間內(nèi)產(chǎn)生大量中斷, 因此在后半部執(zhí)行前有幾個(gè)中斷到達(dá)就不是不尋常的. 驅(qū)動(dòng)必須一直準(zhǔn)備這種可能性并且必須能夠從前半部留下的信息中決定有多少工作要做.

10.3.2.?工作隊(duì)列

回想, 工作隊(duì)列在將來某個(gè)時(shí)候調(diào)用一個(gè)函數(shù), 在一個(gè)特殊工作者進(jìn)程的上下文中. 因?yàn)檫@個(gè)工作隊(duì)列函數(shù)在進(jìn)程上下文運(yùn)行, 它在需要時(shí)能夠睡眠. 但是, 你不能從一個(gè)工作隊(duì)列拷貝數(shù)據(jù)到用戶空間, 除非你使用我們在 15 章演示的高級技術(shù); 工作者進(jìn)程不存取任何其他進(jìn)程的地址空間.

short 驅(qū)動(dòng), 如果設(shè)置 wq 選項(xiàng)為一個(gè)非零值來加載, 為它的后半部處理使用一個(gè)工作隊(duì)列. 它使用系統(tǒng)缺省的工作隊(duì)列, 因此不要求特殊的設(shè)置代碼; 如果你的驅(qū)動(dòng)有特別的運(yùn)行周期要求(或者可能在工作隊(duì)列函數(shù)長時(shí)間睡眠), 你可能需要?jiǎng)?chuàng)建你自己的, 專用的工作隊(duì)列. 我們確實(shí)需要一個(gè) work_struct 結(jié)構(gòu), 它聲明和初始化使用下列:

static struct work_struct short_wq;
/* this line is in short_init() */
INIT_WORK(&short_wq, (void (*)(void *)) short_do_tasklet, NULL);

我們的工作者函數(shù)是 short_do_tasklet, 我們已經(jīng)在前面一節(jié)看到.

當(dāng)使用一個(gè)工作隊(duì)列, short 還建立另一個(gè)中斷處理, 看來如此:

irqreturn_t short_wq_interrupt(int irq, void *dev_id, struct pt_regs *regs)
{
        /* Grab the current time information. */
        do_gettimeofday((struct timeval *) tv_head);
        short_incr_tv(&tv_head);
        /* Queue the bh. Don't worry about multiple enqueueing */
        schedule_work(&short_wq);
        short_wq_count++; /* record that an interrupt arrived */
        return IRQ_HANDLED;
}

如你所見, 中斷處理看來非常象這個(gè) tasklet 版本, 除了它調(diào)用 schedule_work 來安排后半部處理.