在面試中關(guān)于多線程同步，你必須要思考的問(wèn)題 一文中，我們知道glibc的pthread_cond_timedwait底層是用linux futex機(jī)制實(shí)現(xiàn)的。

創(chuàng)新互聯(lián)長(zhǎng)期為上千多家客戶提供的網(wǎng)站建設(shè)服務(wù)，團(tuán)隊(duì)從業(yè)經(jīng)驗(yàn)10年，關(guān)注不同地域、不同群體，并針對(duì)不同對(duì)象提供差異化的產(chǎn)品和服務(wù)；打造開(kāi)放共贏平臺(tái)，與合作伙伴共同營(yíng)造健康的互聯(lián)網(wǎng)生態(tài)環(huán)境。為陽(yáng)西企業(yè)提供專(zhuān)業(yè)的網(wǎng)站設(shè)計(jì)制作、成都網(wǎng)站制作，陽(yáng)西網(wǎng)站改版等技術(shù)服務(wù)。擁有十載豐富建站經(jīng)驗(yàn)和眾多成功案例,為您定制開(kāi)發(fā)。

理想的同步機(jī)制應(yīng)該是沒(méi)有鎖沖突時(shí)在用戶態(tài)利用原子指令就解決問(wèn)題，而需要掛起等待時(shí)再使用內(nèi)核提供的系統(tǒng)調(diào)用進(jìn)行睡眠與喚醒。換句話說(shuō)，在用戶態(tài)的自旋失敗時(shí)，能不能讓進(jìn)程掛起，由持有鎖的線程釋放鎖時(shí)將其喚醒？
如果你沒(méi)有較深入地考慮過(guò)這個(gè)問(wèn)題，很可能想當(dāng)然的認(rèn)為類(lèi)似于這樣就行了（偽代碼）：

void?lock(int?lockval)?{	//trylock是用戶級(jí)的自旋鎖
	while(!trylock(lockval))?{
		wait();//釋放cpu，并將當(dāng)期線程加入等待隊(duì)列，是系統(tǒng)調(diào)用
	}
}boolean?trylock(int?lockval){	int?i=0;?
	//localval=1代表上鎖成功
	while(!compareAndSet(lockval,0,1)){		if(++i>10){			return?false;
		}
	}	return?true;
}void?unlock(int?lockval)?{
	?compareAndSet(lockval,1,0);
	?notify();
}

上述代碼的問(wèn)題是trylock和wait兩個(gè)調(diào)用之間存在一個(gè)窗口：
如果一個(gè)線程trylock失敗，在調(diào)用wait時(shí)持有鎖的線程釋放了鎖，當(dāng)前線程還是會(huì)調(diào)用wait進(jìn)行等待，但之后就沒(méi)有人再喚醒該線程了。

為了解決上述問(wèn)題，linux內(nèi)核引入了futex機(jī)制，futex主要包括等待和喚醒兩個(gè)方法：futex_wait和futex_wake，其定義如下

//uaddr指向一個(gè)地址，val代表這個(gè)地址期待的值，當(dāng)*uaddr==val時(shí)，才會(huì)進(jìn)行waitint?futex_wait(int?*uaddr,?int?val);//喚醒n個(gè)在uaddr指向的鎖變量上掛起等待的進(jìn)程int?futex_wake(int?*uaddr,?int?n);

futex在真正將進(jìn)程掛起之前會(huì)檢查addr指向的地址的值是否等于val，如果不相等則會(huì)立即返回，由用戶態(tài)繼續(xù)trylock。否則將當(dāng)期線程插入到一個(gè)隊(duì)列中去，并掛起。

在關(guān)于同步的一點(diǎn)思考-上文章中對(duì)futex的背景與基本原理有介紹，對(duì)futex不熟悉的人可以先看下。

本文將深入分析futex的實(shí)現(xiàn)，讓讀者對(duì)于鎖的最底層實(shí)現(xiàn)方式有直觀認(rèn)識(shí)，再結(jié)合之前的兩篇文章（關(guān)于同步的一點(diǎn)思考-上和關(guān)于同步的一點(diǎn)思考-下）能對(duì)操作系統(tǒng)的同步機(jī)制有個(gè)全面的理解。

下文中的進(jìn)程一詞包括常規(guī)進(jìn)程與線程。

futex_wait

在看下面的源碼分析前，先思考一個(gè)問(wèn)題：如何確保掛起進(jìn)程時(shí)，val的值是沒(méi)有被其他進(jìn)程修改過(guò)的？

代碼在kernel/futex.c中

static?int?futex_wait(u32?__user?*uaddr,?int?fshared,
		??????u32?val,?ktime_t?*abs_time,?u32?bitset,?int?clockrt){	struct?hrtimer_sleeper?timeout,?*to?=?NULL;
	struct?restart_block?*restart;
	struct?futex_hash_bucket?*hb;
	struct?futex_q?q;
	int?ret;

	...	//設(shè)置hrtimer定時(shí)任務(wù)：在一定時(shí)間(abs_time)后，如果進(jìn)程還沒(méi)被喚醒則喚醒wait的進(jìn)程
	if?(abs_time)?{
	????...
		hrtimer_init_sleeper(to,?current);
		...
	}

retry:	//該函數(shù)中判斷uaddr指向的值是否等于val，以及一些初始化操作
	ret?=?futex_wait_setup(uaddr,?val,?fshared,?&q,?&hb);	//如果val發(fā)生了改變，則直接返回
	if?(ret)		goto?out;	//將當(dāng)前進(jìn)程狀態(tài)改為T(mén)ASK_INTERRUPTIBLE，并插入到futex等待隊(duì)列，然后重新調(diào)度。
	futex_wait_queue_me(hb,?&q,?to);	/*?If?we?were?woken?(and?unqueued),?we?succeeded,?whatever.?*/
	ret?=?0;	//如果unqueue_me成功，則說(shuō)明是超時(shí)觸發(fā)（因?yàn)閒utex_wake喚醒時(shí)，會(huì)將該進(jìn)程移出等待隊(duì)列，所以這里會(huì)失?。?	if?(!unqueue_me(&q))		goto?out_put_key;
	ret?=?-ETIMEDOUT;	if?(to?&&?!to->task)		goto?out_put_key;	/*
	?*?We?expect?signal_pending(current),?but?we?might?be?the
	?*?victim?of?a?spurious?wakeup?as?well.
	?*/
	if?(!signal_pending(current))?{
		put_futex_key(fshared,?&q.key);		goto?retry;
	}

	ret?=?-ERESTARTSYS;	if?(!abs_time)		goto?out_put_key;

	...

out_put_key:
	put_futex_key(fshared,?&q.key);
out:	if?(to)?{		//取消定時(shí)任務(wù)
		hrtimer_cancel(&to->timer);
		destroy_hrtimer_on_stack(&to->timer);
	}	return?ret;
}

在將進(jìn)程阻塞前會(huì)將當(dāng)期進(jìn)程插入到一個(gè)等待隊(duì)列中，需要注意的是這里說(shuō)的等待隊(duì)列其實(shí)是一個(gè)類(lèi)似Java HashMap的結(jié)構(gòu)，全局唯一。

struct?futex_hash_bucket?{
	spinlock_t?lock;	//雙向鏈表
	struct?plist_head?chain;};static?struct?futex_hash_bucket?futex_queues[1<
著重看futex_wait_setup和兩個(gè)函數(shù)futex_wait_queue_me
static?int?futex_wait_setup(u32?__user?*uaddr,?u32?val,?int?fshared,			???struct?futex_q?*q,?struct?futex_hash_bucket?**hb){
	u32?uval;	int?ret;
retry:
	q->key?=?FUTEX_KEY_INIT;	//初始化futex_q
	ret?=?get_futex_key(uaddr,?fshared,?&q->key,?VERIFY_READ);	if?(unlikely(ret?!=?0))		return?ret;

retry_private:	//獲得自旋鎖
	*hb?=?queue_lock(q);	//原子的將uaddr的值設(shè)置到uval中
	ret?=?get_futex_value_locked(&uval,?uaddr);

???...?
	//如果當(dāng)期uaddr指向的值不等于val，即說(shuō)明其他進(jìn)程修改了
	//uaddr指向的值，等待條件不再成立，不用阻塞直接返回。
	if?(uval?!=?val)?{		//釋放鎖
		queue_unlock(q,?*hb);
		ret?=?-EWOULDBLOCK;
	}

???...	return?ret;
}
函數(shù)futex_wait_setup中主要做了兩件事，一是獲得自旋鎖，二是判斷*uaddr是否為預(yù)期值。
static?void?futex_wait_queue_me(struct?futex_hash_bucket?*hb,?struct?futex_q?*q,
				struct?hrtimer_sleeper?*timeout)
{	//設(shè)置進(jìn)程狀態(tài)為T(mén)ASK_INTERRUPTIBLE，cpu調(diào)度時(shí)只會(huì)選擇
	//狀態(tài)為T(mén)ASK_RUNNING的進(jìn)程
	set_current_state(TASK_INTERRUPTIBLE);	//將當(dāng)期進(jìn)程（q封裝）插入到等待隊(duì)列中去，然后釋放自旋鎖
	queue_me(q,?hb);	//啟動(dòng)定時(shí)任務(wù)
	if?(timeout)?{
		hrtimer_start_expires(&timeout->timer,?HRTIMER_MODE_ABS);		if?(!hrtimer_active(&timeout->timer))
			timeout->task?=?NULL;
	}	/*
	?*?If?we?have?been?removed?from?the?hash?list,?then?another?task
	?*?has?tried?to?wake?us,?and?we?can?skip?the?call?to?schedule().
	?*/
	if?(likely(!plist_node_empty(&q->list)))?{		?
		?//如果沒(méi)有設(shè)置過(guò)期時(shí)間?||?設(shè)置了過(guò)期時(shí)間且還沒(méi)過(guò)期
		if?(!timeout?||?timeout->task)			//系統(tǒng)重新進(jìn)行進(jìn)程調(diào)度，這個(gè)時(shí)候cpu會(huì)去執(zhí)行其他進(jìn)程，該進(jìn)程會(huì)阻塞在這里
			schedule();
	}	//走到這里說(shuō)明又被cpu選中運(yùn)行了
	__set_current_state(TASK_RUNNING);
}
futex_wait_queue_me中主要做幾件事：
將當(dāng)期進(jìn)程插入到等待隊(duì)列
啟動(dòng)定時(shí)任務(wù)
重新調(diào)度進(jìn)程
如何保證條件與等待之間的原子性
在futex_wait_setup方法中會(huì)加自旋鎖；在futex_wait_queue_me中將狀態(tài)設(shè)置為TASK_INTERRUPTIBLE，調(diào)用queue_me將當(dāng)期線程插入到等待隊(duì)列中，然后才釋放自旋鎖。也就是說(shuō)檢查uaddr的值的過(guò)程跟進(jìn)程掛起的過(guò)程放在同一個(gè)臨界區(qū)中。當(dāng)釋放自旋鎖后，這時(shí)再更改addr地址的值已經(jīng)沒(méi)有關(guān)系了，因?yàn)楫?dāng)期進(jìn)程已經(jīng)加入到等待隊(duì)列中，能被wake喚醒，不會(huì)出現(xiàn)本文開(kāi)頭提到的沒(méi)人喚醒的問(wèn)題。
futex_wait小結(jié)
總結(jié)下futex_wait流程：
加自旋鎖
檢測(cè)*uaddr是否等于val，如果不相等則會(huì)立即返回
將進(jìn)程狀態(tài)設(shè)置為TASK_INTERRUPTIBLE
將當(dāng)期進(jìn)程插入到等待隊(duì)列中
釋放自旋鎖
創(chuàng)建定時(shí)任務(wù)：當(dāng)超過(guò)一定時(shí)間還沒(méi)被喚醒時(shí)，將進(jìn)程喚醒
掛起當(dāng)前進(jìn)程
futex_wake
futex_wake
static?int?futex_wake(u32?__user?*uaddr,?int?fshared,?int?nr_wake,?u32?bitset){	struct?futex_hash_bucket?*hb;
	struct?futex_q?*this,?*next;
	struct?plist_head?*head;
	union?futex_key?key?=?FUTEX_KEY_INIT;	int?ret;

	...	//根據(jù)uaddr的值填充&key的內(nèi)容
	ret?=?get_futex_key(uaddr,?fshared,?&key,?VERIFY_READ);	if?(unlikely(ret?!=?0))		goto?out;	//根據(jù)&key獲得對(duì)應(yīng)uaddr所在的futex_hash_bucket
	hb?=?hash_futex(&key);	//對(duì)該hb加自旋鎖
	spin_lock(&hb->lock);
	head?=?&hb->chain;	//遍歷該hb的鏈表，注意鏈表中存儲(chǔ)的節(jié)點(diǎn)是plist_node類(lèi)型，而而這里的this卻是futex_q類(lèi)型，這種類(lèi)型轉(zhuǎn)換是通過(guò)c中的container_of機(jī)制實(shí)現(xiàn)的
	plist_for_each_entry_safe(this,?next,?head,?list)?{		if?(match_futex?(&this->key,?&key))?{
			...			//喚醒對(duì)應(yīng)進(jìn)程
			wake_futex(this);			if?(++ret?>=?nr_wake)				break;
		}
	}	//釋放自旋鎖
	spin_unlock(&hb->lock);
	put_futex_key(fshared,?&key);
out:	return?ret;
}
futex_wake流程如下：
找到uaddr對(duì)應(yīng)的futex_hash_bucket，即代碼中的hb
對(duì)hb加自旋鎖
遍歷fb的鏈表，找到uaddr對(duì)應(yīng)的節(jié)點(diǎn)
調(diào)用wake_futex喚起等待的進(jìn)程
釋放自旋鎖
wake_futex中將制定進(jìn)程狀態(tài)設(shè)置為TASK_RUNNING并加入到系統(tǒng)調(diào)度列表中，同時(shí)將進(jìn)程從futex的等待隊(duì)列中移除掉，具體代碼就不分析了，有興趣的可以自行研究。
End
Java中的ReentrantLock,Object.wait和Thread.sleep等等底層都是用futex進(jìn)行線程同步，理解futex的實(shí)現(xiàn)能幫助你更好的理解與使用這些上層的同步機(jī)制。另外因篇幅與精力有限，涉及到進(jìn)程調(diào)度的相關(guān)內(nèi)容沒(méi)有具體分析，不過(guò)并不妨礙理解文章內(nèi)容。
            
            
                        

            網(wǎng)站名稱(chēng)：linux內(nèi)核級(jí)同步機(jī)制--futex            

            分享網(wǎng)址：http://weahome.cn/article/jpcdsg.html