瑞芯微RK3568芯片是一款定位中高端的通用型SOC，采用22nm制程工藝，搭載一顆四核Cortex-A55處理器和Mali G52 2EE圖形處理器。RK3568支持4K解碼和1080P編碼，支持SATA/PCIE/USB3.0外圍接口。RK3568內(nèi)置獨(dú)立NPU，可用于輕量級人工智能應(yīng)用。RK3568支持安卓11和linux系統(tǒng)，主要面向物聯(lián)網(wǎng)網(wǎng)關(guān)、NVR存儲、工控平板、工業(yè)檢測、工控盒、卡拉OK、云終端、車載中控等行業(yè)。

?

迅為RK568郵票孔版開發(fā)板：
還在為工控機(jī)性能不足而煩惱？
還在為開發(fā)周期長而焦慮？
迅為RK568郵票孔版開發(fā)板，為您提供一站式解決方案！

強(qiáng)勁性能，賦能工業(yè)未來：
搭載瑞芯微RK568高性能處理器，采用先進(jìn)制程工藝，提供澎湃算力，輕松應(yīng)對復(fù)雜工業(yè)場景。
支持多種操作系統(tǒng)，如Android、Linux等，滿足不同應(yīng)用需求。
豐富接口，擴(kuò)展性強(qiáng)，可連接多種工業(yè)設(shè)備，構(gòu)建穩(wěn)定可靠的工業(yè)控制系統(tǒng)。

郵票孔設(shè)計(jì)，靈活易用：
采用郵票孔連接方式，方便快捷，節(jié)省空間，易于集成到各種工控設(shè)備中。
提供配套外殼，防護(hù)等級高，適應(yīng)各種惡劣工業(yè)環(huán)境。
提供完善的技術(shù)支持和開發(fā)資料，助您快速上手，縮短開發(fā)周期。

廣泛應(yīng)用于工業(yè)自動化控制、機(jī)器視覺、智能網(wǎng)關(guān)、工業(yè)機(jī)器人、物聯(lián)網(wǎng)終端等。

【公眾號】迅為電子

第19章 并發(fā)與競爭實(shí)驗(yàn)

在前面章節(jié)的學(xué)習(xí)中，相信大家已經(jīng)對用戶空間與內(nèi)核空間數(shù)據(jù)傳遞進(jìn)行了實(shí)驗(yàn)，假如要傳遞的數(shù)據(jù)被存放在了全局變量，該數(shù)據(jù)就可以作為共享資源被多個任務(wù)共同讀寫，從而造成數(shù)據(jù)的錯誤傳輸，多個程序同時訪問一個共享資源產(chǎn)生的問題就叫做競爭。競爭產(chǎn)生的根本原因就是Linux系統(tǒng)的并發(fā)訪問。

在本章節(jié)中首先會對并發(fā)與并行的概念進(jìn)行講解，隨后對競爭產(chǎn)生的原因進(jìn)行總結(jié)，最后以一個實(shí)際的競爭實(shí)驗(yàn)加深大家的理解。下面就讓我們開始本章節(jié)的學(xué)習(xí)吧。

19.1并發(fā)與競爭

19.1.1并發(fā)

早期計(jì)算機(jī)大多只有一個CPU核心，一個CPU在同一時間只能執(zhí)行一個任務(wù)，當(dāng)系統(tǒng)中有多個任務(wù)等待執(zhí)行時，CPU只能執(zhí)行完一個再執(zhí)行下一個。而計(jì)算機(jī)的很多指令會涉及I/O操作，執(zhí)行速度遠(yuǎn)遠(yuǎn)低于CPU內(nèi)高速存儲器的存取速度，這就導(dǎo)致CPU經(jīng)常處于空閑狀態(tài)，只能等待I/O操作完成后才能繼續(xù)執(zhí)行后面的指令。為了提高CPU利用率，減少等待時間，提出了CPU并發(fā)工作理論。

所謂并發(fā)，就是通過算法將CPU資源合理地分配給多個任務(wù)，當(dāng)一個任務(wù)執(zhí)行I/O操作時，CPU可以轉(zhuǎn)而執(zhí)行其它的任務(wù)，等到I/O操作完成以后，或者新的任務(wù)遇到I/O操作時，CPU再回到原來的任務(wù)繼續(xù)執(zhí)行。

下圖（圖19-1）展示了兩個任務(wù)并發(fā)執(zhí)行的過程（為了容易理解，這里以兩個任務(wù)并發(fā)執(zhí)行為例，當(dāng)然一個CPU核心并不僅僅只能兩個任務(wù)并發(fā)）：

雖然CPU在同一時刻只能執(zhí)行一個任務(wù)，但是通過將CPU的使用權(quán)在恰當(dāng)?shù)臅r機(jī)分配給不同的任務(wù)，使得多個任務(wù)看起來是一起執(zhí)行的（CPU的執(zhí)行速度極快，多任務(wù)切換的時間也極短）。

至此關(guān)于并發(fā)的概念就講解完成了。

19.1.2并行

并發(fā)是針對單核CPU提出的，而并行則是針對多核CPU提出的。和單核CPU不同，多核CPU真正實(shí)現(xiàn)了“同時執(zhí)行多個任務(wù)”。多核CPU的每個核心都可以獨(dú)立地執(zhí)行一個任務(wù)，而且多個核心之間不會相互干擾。在不同核心上執(zhí)行的多個任務(wù)，是真正地同時運(yùn)行，這種狀態(tài)就叫做并行。雙核CPU的工作狀態(tài)如下圖（圖19-2）所示：

雙核CPU執(zhí)行兩個任務(wù)時，每個核心各自執(zhí)行一個任務(wù)，和單核CPU在兩個任務(wù)之間不斷切換相比，它的執(zhí)行效率更高。

至此對于并行的概念就講解完成了。

19.1.3并發(fā)+并行

在并行的工作狀態(tài)中，兩個CPU分別執(zhí)行兩個任務(wù)，是一種理想狀態(tài)。但是在實(shí)際場景中，處于運(yùn)行狀態(tài)的任務(wù)是非常多的，以實(shí)際辦公電腦為例，windows系統(tǒng)在開機(jī)之后會運(yùn)行幾十個任務(wù)，而CPU往往只有4核、8核等，遠(yuǎn)遠(yuǎn)低于任務(wù)的數(shù)量，這個時候就會同時存在并發(fā)和并行兩種情況，即所有核心在并行工作的同時，每個核心還要并發(fā)工作。

例如一個雙核 CPU要執(zhí)行四個任務(wù)，它的工作狀態(tài)如下圖（圖19-3）所示：

為了容易理解，這里是以兩個任務(wù)并發(fā)執(zhí)行為例，當(dāng)然一個CPU核心并不僅僅只能兩個任務(wù)并發(fā)，并發(fā)任務(wù)的數(shù)量和操作系統(tǒng)的分配方式、以及每個任務(wù)的工作狀態(tài)有關(guān)系。

至此，對于并發(fā)+并行的概念講解就結(jié)束了。

并發(fā)可以看作是并行的理想狀態(tài)，為了便于講解和避免產(chǎn)生歧義，之后的章節(jié)無論是并發(fā)還是并行，都會統(tǒng)稱為并發(fā)。

19.1.4競爭

并發(fā)可能會造成多個程序同時訪問一個共享資源，這時候由并發(fā)同時訪問一個共享資源產(chǎn)生的問題就叫做競爭。

競爭產(chǎn)生的原因如下所示：

（1）多線程的并發(fā)訪問。由于Linux是多任務(wù)操作系統(tǒng)，所以多線程訪問是競爭產(chǎn)生的基本原因。

（2）中斷程序的并發(fā)訪問。中斷任務(wù)產(chǎn)生后，CPU會立刻停止當(dāng)前工作，從而去執(zhí)行中斷中的任務(wù)，如果中斷任務(wù)對共享資源進(jìn)行了修改，就會產(chǎn)生競爭。

（3）搶占式并發(fā)訪問。linux2.6及更高版本引入了搶占式內(nèi)核，高優(yōu)先級的任務(wù)可以打斷低優(yōu)先級的任務(wù)。在線程訪問共享資源的時候，另一個線程打斷了現(xiàn)在正在訪問共享資源的線程同時也對共享資源進(jìn)行操作，從而造成了競爭。

（4）多處理器(SMP）并發(fā)訪問。多核處理器之間存在核間并發(fā)訪問。

19.1.5共享資源的保護(hù)

競爭是由并發(fā)訪問同一個共享資源產(chǎn)生的。為了防止“競爭”的產(chǎn)生就要對共享資源進(jìn)行保護(hù)，這里提到的共享資源又是什么呢？

以實(shí)際生活中的共享資源為例，可以是公共電話，也可以是共享單車、共享充電寶等公共物品，以上都屬于共享資源的范疇，以公共電話為例，每個人都可以對它進(jìn)行使用，但在同一時間內(nèi)只能由一個人進(jìn)行使用，如果兩個人都要對電話進(jìn)行使用，則產(chǎn)生了競爭。而在實(shí)際的驅(qū)動的代碼中，共享資源可以是全局變量，也可以是驅(qū)動中的設(shè)備結(jié)構(gòu)體等，需要根據(jù)具體的驅(qū)動程序來進(jìn)行分析。在下一小節(jié)的實(shí)驗(yàn)中，會以全局變量為例，進(jìn)行并發(fā)與競爭實(shí)驗(yàn)。

19.2實(shí)驗(yàn)程序的編寫

19.2.1驅(qū)動程序編寫

本實(shí)驗(yàn)對應(yīng)的網(wǎng)盤路徑為：iTOP-RK3568開發(fā)板【底板V1.7版本】\03_【iTOP-RK3568開發(fā)板】指南教程\02_Linux驅(qū)動配套資料\04_Linux驅(qū)動例程\14\module。

本實(shí)驗(yàn)將編寫并發(fā)與競爭的驅(qū)動代碼，首先完善字符設(shè)備驅(qū)動框架，然后通過copy_from_user(…)函數(shù)接收用戶空間傳遞到內(nèi)核空間的數(shù)據(jù)并進(jìn)行判斷，如果接收到的字符串?dāng)?shù)據(jù)為“topeet”會在睡眠4秒鐘后打印接收到的數(shù)據(jù)，如果接收到的字符串?dāng)?shù)據(jù)為“itop”會在睡眠2秒鐘后打印接收到的數(shù)據(jù)。

編寫完成的example.c代碼如下所示

#include

#include

#include

#include

#include

#include

#include

static int open_test(struct inode *inode,struct file *file)

{

printk("\nthis is open_test \n");

return 0;

}

static ssize_t read_test(struct file *file,char __user *ubuf,size_t len,loff_t *off)

{

int ret;

char kbuf[10] = "topeet";//定義char類型字符串變量kbuf

printk("\nthis is read_test \n");

ret = copy_to_user(ubuf,kbuf,strlen(kbuf));//使用copy_to_user接收用戶空間傳遞的數(shù)據(jù)

if (ret != 0){

printk("copy_to_user is error \n");

}

printk("copy_to_user is ok \n");

return 0;

}

static char kbuf[10] = {0};//定義char類型字符串全局變量kbuf

static ssize_t write_test(struct file *file,const char __user *ubuf,size_t len,loff_t *off)

{

int ret;

ret = copy_from_user(kbuf,ubuf,len);//使用copy_from_user接收用戶空間傳遞的數(shù)據(jù)

if (ret != 0){

printk("copy_from_user is error\n");

}

if(strcmp(kbuf,"topeet") == 0 ){//如果傳遞的kbuf是topeet就睡眠四秒鐘

ssleep(4);

}

else if(strcmp(kbuf,"itop") == 0){//如果傳遞的kbuf是itop就睡眠兩秒鐘

ssleep(2);

}

printk("copy_from_user buf is %s \n",kbuf);

return 0;

}

static int release_test(struct inode *inode,struct file *file)

{

//printk("\nthis is release_test \n");

return 0;

}

struct chrdev_test {

dev_t dev_num;//定義dev_t類型變量dev_num來表示設(shè)備號

int major,minor;//定義int類型的主設(shè)備號major和次設(shè)備號minor

struct cdev cdev_test;//定義struct cdev類型結(jié)構(gòu)體變量cdev_test，表示要注冊的字符設(shè)備

struct class *class_test;//定于struct class *類型結(jié)構(gòu)體變量class_test，表示要創(chuàng)建的類

};

struct chrdev_test dev1;//創(chuàng)建chrdev_test類型的

struct file_operations fops_test = {

.owner = THIS_MODULE,//將owner字段指向本模塊，可以避免在模塊的操作正在被使用時卸載該模塊

.open = open_test,//將open字段指向open_test(...)函數(shù)

.read = read_test,//將read字段指向read_test(...)函數(shù)

.write = write_test,//將write字段指向write_test(...)函數(shù)

.release = release_test,//將release字段指向release_test(...)函數(shù)

};

static int __init atomic_init(void)

{

if(alloc_chrdev_region(&dev1.dev_num,0,1,"chrdev_name") < 0 ){//自動獲取設(shè)備號，設(shè)備名chrdev_name

printk("alloc_chrdev_region is error \n");

}

printk("alloc_chrdev_region is ok \n");

dev1.major = MAJOR(dev1.dev_num);//使用MAJOR()函數(shù)獲取主設(shè)備號

dev1.minor = MINOR(dev1.dev_num);//使用MINOR()函數(shù)獲取次設(shè)備號

printk("major is %d,minor is %d\n",dev1.major,dev1.minor);

cdev_init(&dev1.cdev_test,&fops_test);//使用cdev_init()函數(shù)初始化cdev_test結(jié)構(gòu)體，并鏈接到fops_test結(jié)構(gòu)體

dev1.cdev_test.owner = THIS_MODULE;//將owner字段指向本模塊，可以避免在模塊的操作正在被使用時卸載該模塊

cdev_add(&dev1.cdev_test,dev1.dev_num,1);//使用cdev_add()函數(shù)進(jìn)行字符設(shè)備的添加

dev1.class_test = class_create(THIS_MODULE,"class_test");//使用class_create進(jìn)行類的創(chuàng)建，類名稱為class_test

device_create(dev1.class_test,0,dev1.dev_num,0,"device_test");//使用device_create進(jìn)行設(shè)備的創(chuàng)建，設(shè)備名稱為device_test

return 0;

}

static void __exit atomic_exit(void)

{

device_destroy(dev1.class_test,dev1.dev_num);//刪除創(chuàng)建的設(shè)備

class_destroy(dev1.class_test);//刪除創(chuàng)建的類

cdev_del(&dev1.cdev_test);//刪除添加的字符設(shè)備cdev_test

unregister_chrdev_region(dev1.dev_num,1);//釋放字符設(shè)備所申請的設(shè)備號

printk("module exit \n");

}

module_init(atomic_init);

module_exit(atomic_exit)

MODULE_LICENSE("GPL v2");

MODULE_AUTHOR("topeet");

對于重要邏輯部分已經(jīng)加粗，后續(xù)章節(jié)的實(shí)驗(yàn)都是對上述并發(fā)與競爭實(shí)驗(yàn)的改進(jìn)，以不同的方式來避免競爭的產(chǎn)生。

19.2.2編寫測試APP

本實(shí)驗(yàn)應(yīng)用程序?qū)?yīng)的網(wǎng)盤路徑為：iTOP-RK3568開發(fā)板【底板V1.7版本】\03_【iTOP-RK3568開發(fā)板】指南教程\02_Linux驅(qū)動配套資料\04_Linux驅(qū)動例程\14\app。

本測試app較為簡單，需要輸入兩個參數(shù)，第一個參數(shù)為對應(yīng)的設(shè)備節(jié)點(diǎn)，第二個參數(shù)為“topeet”或者“itop”，分別代表向設(shè)備寫入的數(shù)據(jù)，編寫完成的應(yīng)用程序app.c內(nèi)容如下所示：

#include

#include

#include

#include

#include

#include

int main(int argc, char *argv[])

{

int fd;//定義int類型的文件描述符

char str1[10] = {0};//定義讀取緩沖區(qū)str1

fd = open(argv[1],O_RDWR);//調(diào)用open函數(shù)，打開輸入的第一個參數(shù)文件，權(quán)限為可讀可寫

if(fd < 0 ){

printf("file open failed \n");

return -1;

}

/*如果第二個參數(shù)為topeet，條件成立，調(diào)用write函數(shù)，寫入topeet*/

if (strcmp(argv[2],"topeet") == 0 ){

write(fd,"topeet",10);

}

/*如果第二個參數(shù)為itop，條件成立，調(diào)用write函數(shù)，寫入itop*/

else if (strcmp(argv[2],"itop") == 0 ){

write(fd,"itop",10);

}

close(fd);

return 0;

}

19.3運(yùn)行測試

19.3.1編譯驅(qū)動程序

在上一小節(jié)中的example.c代碼同一目錄下創(chuàng)建Makefile文件，Makefile文件內(nèi)容如下所示：

export ARCH=arm64#設(shè)置平臺架構(gòu)

export CROSS_COMPILE=aarch64-linux-gnu-#交叉編譯器前綴

obj-m += example.o #此處要和你的驅(qū)動源文件同名

KDIR :=/home/topeet/Linux/linux_sdk/kernel #這里是你的內(nèi)核目錄

PWD ?= $(shell pwd)

all:

make -C $(KDIR) M=$(PWD) modules #make操作

clean:

make -C $(KDIR) M=$(PWD) clean #make clean操作

對于Makefile的內(nèi)容注釋已在上圖添加，保存退出之后，來到存放example.c和Makefile文件目錄下，如下圖所示：

然后使用命令“make”進(jìn)行驅(qū)動的編譯，編譯完成如下圖（圖19-5）所示：

編譯完生成example.ko目標(biāo)文件，如下圖（圖19-6）所示：

至此驅(qū)動模塊就編譯成功了，下面進(jìn)行應(yīng)用程序的編譯。

19.3.2編譯應(yīng)用程序

來到應(yīng)用程序app.c文件的存放路徑如下圖（圖19-7）所示：

然后使用以下命令對app.c進(jìn)行交叉編譯，編譯完成如下圖（圖19-8）所示：

1

aarch64-linux-gnu-gcc -o app app.c -static

生成的app文件就是之后放在開發(fā)板上運(yùn)行的可執(zhí)行文件，至此應(yīng)用程序的編譯就完成了。

19.3.3運(yùn)行測試

開發(fā)板啟動之后，使用以下命令進(jìn)行驅(qū)動模塊的加載，如下圖（圖19-9）所示：

1

insmod example.ko

可以看到申請的主設(shè)備號和次設(shè)備號就被打印了出來，然后使用以下代碼對自動生成的設(shè)備節(jié)點(diǎn)device_test進(jìn)行查看，如下圖（圖19-10）所示：

1

ls /dev/device_test

可以看到device_test節(jié)點(diǎn)已經(jīng)被自動創(chuàng)建了，然后使用以下命令運(yùn)行測試app，運(yùn)行結(jié)果如下圖（圖19-11）所示：

1

./app /dev/device_test topeet

可以看到傳遞的buf值為topeet，然后輸入以下命令在后臺運(yùn)行兩個app，來進(jìn)行競爭測試，運(yùn)行結(jié)果如下圖（圖19-12）所示：

./app /dev/device_test topeet &

./app /dev/device_test itop &

在不存在競爭的情況下，傳遞的兩個字符串?dāng)?shù)據(jù)應(yīng)該是topeet和itop，而在上圖中的打印信息為兩個itop，原因是第二個app應(yīng)用程序運(yùn)行之后對共享資源進(jìn)行了修改，兩個app應(yīng)用程序就產(chǎn)生了競爭關(guān)系，會在之后的章節(jié)中使用不同的方法對上述驅(qū)動程序進(jìn)行改進(jìn)，從而避免競爭的產(chǎn)生。

最后可以使用以下命令進(jìn)行驅(qū)動的卸載，如下圖（圖19-13）所示：

1

rmmod example.ko

至此，并發(fā)與競爭的實(shí)驗(yàn)就完成了。