10.3. 前和后半部

10.3.?前和后半部

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

Linux (許多其他系統(tǒng)一起)解決這個(gè)問(wèn)題通過(guò)將中斷處理分為 2 半. 所謂的前半部是實(shí)際響應(yīng)中斷的函數(shù) -- 你使用 request_irq 注冊(cè)的那個(gè). 后半部是由前半部調(diào)度來(lái)延后執(zhí)行的函數(shù), 在一個(gè)更安全的時(shí)間. 最大的不同在前半部處理和后半部之間是所有的中斷在后半部執(zhí)行時(shí)都使能 -- 這就是為什么它在一個(gè)更安全時(shí)間運(yùn)行. 在典型的場(chǎng)景中, 前半部保存設(shè)備數(shù)據(jù)到一個(gè)設(shè)備特定的緩存, 調(diào)度它的后半部, 并且退出: 這個(gè)操作非?？? 后半部接著進(jìn)行任何其他需要的工作, 例如喚醒進(jìn)程, 啟動(dòng)另一個(gè) I/O 操作, 等等. 這種設(shè)置允許前半部來(lái)服務(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ī)制可用來(lái)實(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ì)列處理者來(lái)進(jìn)行中斷處理. 在這個(gè)情況下, 前半部快速地執(zhí)行; 它簡(jiǎn)單地記住當(dāng)前時(shí)間并且調(diào)度后半部處理. 后半部接著負(fù)責(zé)將時(shí)間編碼并且喚醒任何可能在等待數(shù)據(jù)的用戶(hù)進(jìn)程.

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

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

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

tasklet 必須使用 DECLARE_TASKLET 宏來(lái)聲明:

DECLARE_TASKLET(name, function, data);

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

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

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

函數(shù) tasklet_schedule 用來(lái)調(diào)度一個(gè) tasklet 運(yùn)行. 如果 short 使用 tasklet=1 來(lái)加載, 它安裝一個(gè)不同的中斷處理來(lái)保存數(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í)行. 如同前面提過(guò), 這個(gè)函數(shù)進(jìn)行處理中斷的大量工作; 它看來(lái)如此:

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)用以來(lái)有多少中斷到達(dá). 一個(gè)如 short 一樣的設(shè)備能夠在短時(shí)間內(nèi)產(chǎn)生大量中斷, 因此在后半部執(zhí)行前有幾個(gè)中斷到達(dá)就不是不尋常的. 驅(qū)動(dòng)必須一直準(zhǔn)備這種可能性并且必須能夠從前半部留下的信息中決定有多少工作要做.

10.3.2.?工作隊(duì)列

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

short 驅(qū)動(dòng), 如果設(shè)置 wq 選項(xiàng)為一個(gè)非零值來(lái)加載, 為它的后半部處理使用一個(gè)工作隊(duì)列. 它使用系統(tǒng)缺省的工作隊(duì)列, 因此不要求特殊的設(shè)置代碼; 如果你的驅(qū)動(dòng)有特別的運(yùn)行周期要求(或者可能在工作隊(duì)列函數(shù)長(zhǎng)時(shí)間睡眠), 你可能需要?jiǎng)?chuàng)建你自己的, 專(zhuān)用的工作隊(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è)中斷處理, 看來(lái)如此:

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;
}

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