2016年2月9日星期二

SH_DIV

#define SH_DIV(NOM,DEN,LSH) (   (((NOM) / (DEN)) << (LSH))              \
                             + ((((NOM) % (DEN)) << (LSH)) + (DEN) / 2) / (DEN))


NOM = NOM / DEN + NOM % DEN
 

(NOM << LSH) / DEN

= (NOM / DEN + NOM % DEN) << LSH / DEN

= (NOM / DEN) << LSH

一个描述paxos算法的恰当的例子

最近看了paxos算法。从发明者本人开始,到网上各种介绍算法的文章,都是在通过各种例子说明这个算法。但我觉得,似乎大部分的例子在让人理解的时候都会引入歧义。比如发明者本人以一个虚构的希腊城邦为例,我看了例子后很长时间都没有搞明白一点:当提案通过之后,是不是还能够继续修订提案?比如这次会议决定了拦路抢劫判10年,通过之后是不是可以再重新提案修改为判20年?当我最终弄明白这个算法之后,自己想了下面一个例子。我认为,我这个例子可以避免误解,让人直接明白paxos算法究竟要解决的是什么问题。

假设这样的一个场景:你和另外四个人被一个性格古怪的恐怖分子劫持了,他要求和你们五个人做一个游戏。游戏规则是,他会把你们五个人分别放进五个房间,每个房间都有一张桌子和一个窗户。你们互相之间都看不到其他人的房间或窗户。桌子上放着三样东西,这三样东西各不相同,但每人桌子上的东西是相同的。比如,你的桌子上是铅笔,橡皮,绳子,其他人的桌子上也都是铅笔,橡皮,绳子。但摆放的位置可能不同。你们每人需要从桌子上选择一件东西扔到窗户外面。如果有三个或更多的人将同样的东西扔到了窗户外面,就算你们赢了,你们每人会得到1千万元的奖励,如果没有三个人扔出同样的东西,比如两个人扔了橡皮,两个人仍了铅笔,一个人扔了绳子,那么你们就输了。输了就全都被枪毙。在游戏分出输赢之前,你们五个人会被一直关在屋子里,不许出来,你们有吃有喝,不会饿死。

接下来是关键的条件:你们每人都可以拿着一部手机,用来互相通信。这个手机的功能被做了如下限制:不能打电话,只能发短信,且每个人只能发送短信给其他四个人。手机通讯录上记录了其他四个人的联系方式。但这个手机短信的收发功能不大可靠,你发送的一条短信,可能丢失,最终没有人收到,也可能延迟很久才被人收到。而且,还有一点不利的条件,你们中最多可能有两个人完全无法和别人联络,也就是说,可能有0到2个人,发出去的短信别人永远无法接到,也永远受不到别人的短信。

前提条件已经讲完了,实际上问题就是,你们五个人进入屋子后,要通过手机短信协商出把哪件东西扔出窗外,并且这个手机短信的传输信道很不靠谱,可能会丢失短信,或者让短信延迟很长时间才会收到。甚至,可能会有1到2部手机从一开始或者从某个时间点开始,永远都联系不上其他三部手机。现在,在分别被关进屋子之前,你们五个人被聚集在一起,可以一起商量一个办法,商量进入屋子后,如何才能让至少三个人把同一件东西扔出窗外。

我尝试分别问了两个人,这两个人的第一反应是相似的,都提出了这样的办法:选定一个人,由这个人发短信给其他人,告诉其他人该扔哪件东西。我反驳说:按照前面提到的条件,最多可能有两个人可其他人完全无法通讯,如果恰好是开始选出的人,岂不是永远都无法做出决定了?这两个人接下来的反应也大同小异:那就给五个人编号,分别为1,2,3,4,5,如果1发出去的消息别人收不到,那就由2发,如果2发出去的别人还收不到,那就由3发。我会反问:由于消息可能会延迟也可能会丢失,假设1和2的消息都延误了很久,3开始发消息了,结果最终只有5收到了1的消息,1发给其他人的消息都丢失了,4收到了3的消息,3发给其他人的消息也都丢失了。而2没有收到任何人的消息,2发出去的消息也都丢失了。这时1,5扔出去了一样的东西,3,4扔出去了一样的东西,2自己扔出去了一件东西,结果没有3个人扔一样的东西,你们就输了。所以,这样的思路是行不通的。

Paxos算法是一个可行的思路。按照paxos算法,我们可以为每个人都定下这样的规则:

进入屋子之后,随机等待一段时间,开始和别人联系,联系的内容就是提议将某件物品扔出窗外。提议分两步进行,第一步是发送短信联系5个人中的任意3个人(可以包括自己),短信包含如下内容:
我是xxx,我想进行提议,本次提议的序列号SN
我们把这条短信叫做“短信1”
注意,这条短信中不包括提议仍哪件东西的具体内容,只表明你想进行提议,SN是一个全局唯一的序列号,并且单调递增。这里我们可以先假设SN就是时间戳,每个人打算提议时就记下当前的时间,把时间戳发出去。当然,题目中并没有说关在屋子里的人可以看时间,而且在两个人同时进行提议的时候,时间戳也可能相同,不能保证全局唯一性。在讲完paxos算法后,我会说明如何在不借助任何外部工具的情况下构建出这样一个全局唯一并且单调递增的SN号。但目前,为了简单起见,我们先假设SN号就是时间戳。

大家都是随机等待一段时间开始发“短信1”的,随机的目的是为了让大家把发短信的时间岔开。但有可能两个人在很接近的时间内发送了“短信1”给同一个人,这样有人就接连收到了两条“短信1”,我们给接收到短信的人指定这样一个规则:他如果同时收到了多条“短信1”,就比较SN号,只回复SN号最大的那个人发来的短信,其他SN号比较小的,就都不理会。

对于回复“短信1”的人,如果他之前还没同意过任何提议,则回复内容如下:
我是xxx,我收到序列号为SN的提议请求,我同意你进行提议,并且,我之前没有同意过任何提议。
如果回复“短信1”的人之前同意过其他人的提议(我们稍候会介绍如何同意其他人的提议),那么他选出所有他同意的提议里面,SN号最大的那个提议,并回复内容如下:
我是xxx,我收到序列号为SN的提议请求,我同意你进行提议,并且,我之前同意过的最新的提议为"扔出xx",该提议的序列号为SN1。

对于发送“短信1”的人,他等待一段固定的时间,比如说十分钟,如果在十分中内没有收到3个人对“短信1”的回复,比如只收到了2个人的回复,或者也可能一个回复都没有收到,那么他就再随机等待一段时间,然后再尝试找3个人发“短信1”,每次找的3个人最好都有变化,别总是找固定的3个人,因为如一开始的假设,可能有最多两个人和其他人是完全失联的,如果总找完全失联的人,就可能永远集不齐3个回复。

当某个人在某次发了“短信1”之后,如果收到了3个人对“短信1”的回复,那么他开始发送“短信2”。“短信2”的内容视他收到的对“短信1”的回复而定。假如他收到的3个对“短信1”的回复都说“我之前没有同意过任何提议”,那么,他就任选一个桌子上的物品,比如说橡皮,然后发送“短信2”的内容如下:
我提议把橡皮扔出窗外,我的序列号是SN。
这个SN依然是发送短信1时使用的序列号,不是新生成一个序列号。

如果他收到的三个对短信1的回复中有一个或多个人回复说“我之前同意过的最新的提议为"扔出xx",该提议的序列号为SN1”,那么他就从这些回复中选出序列号最大的那个提议,看看那个提议是建议扔出什么东西,然后他也提议仍同样的东西。比如,他看到3条对“短信1”的回复里面,序列号最大的提议是扔出铅笔,那么他就发送“短信2”的内容如下:
我提议把铅笔扔出窗外,我的序列号是SN。
注意,扔出铅笔是对“短信1”的回复里面序列号最大的那个提议的内容,但序列号SN依然是他自己发出“短信1”时确定的序列号。

他把“短信2”发送给任意3个人,可以和之前发给“短信1”的三个人相同,也可以不相同。然后他等待其他人的回复,如果他收到了3条回复,并且内容如下:
同意你的序列号为SN的提议
那么他就把他提议的东西扔出窗外。如果等了一段固定的时间,没有集齐3个回复,那么他就再随机等待一段时间,然后重新从“短信1”开始发送。

接下来我们说收到“短信2”的人如何回应。如果一个人收到了“短信2”,他首先看看有没有收到序列号更大的"短信1"或者“短信2”,如果有的话,就回复序列号最大的那条“短信1”或者“短信2”,忽略这条最新收到的“短信2”。注意,由于发送信息可能会有延迟,最新收到的短信的序列号不一定是最大的,因此,他可能会忽略最新收到的短信。如果这个人发现他收到的这条“短信2”是他所有收到的短信里面序列号最大的(或者并列最大的),那么他就回复说:
同意你的序列号为SN的的提议
关于并列最大的序列号,我们说序列号全局唯一,并且单调递增,但同一个序列号即要被用于”短信1“也要被用于”短信2“,可能这个人收到过序列号为SN1的“短信1”,然后他又收到了序列号为SN1的“短信2”,因此,可能出现并列最大的序列号。

当收到“短信2”的人做出了这样的回复之后,他就记录下这是他最新同意的提议,如果接下来他又收到序列号更新的“短信1”的时候,他就回复说他同意了这个提议。当他回复了“短信2”之后,他可能又收到一个序列号为SN1的“短信2”,这个SN1比SN更大,这时,他也会做如下回复:
同意你的序列号为SN1的的提议
同时,他也会记录下,SN1以及其对应的提议是他同意过的最新的提议,再收到“短信1”的时候,他会把SN1以及SN1对应的提议返回给发送“短信1”的人。

每个人都按照上述规则发送“短信1”和“短信2”,并且处理别人发来的“短信1”和“短信2”。最终,这5个人会达成一致,把同一件东西扔出窗外,或者,假设有1或2个人处于完全失联的状态的话,其余的人会达成一致,把同一件东西扔出窗外。

算法的正确性有严格的证明,可以在网上搜到。我写这篇文章的目的是方便大家理解paxos算法,对于算法的正确性,我就不作证明了。下面我们通过具体的例子来感受一下算法的运行:

按照本文开头假设的场景,有A,B,C,D,E五个人,他们被分别关进了5个屋子,每个屋子里面的桌子上,都摆放着铅笔,橡皮,绳子。现在,他们要通过可能会丢包/延迟的手机短信,对把哪个东西扔出窗外这个问题达成一致。

进入屋子之后,过了一段时间,A决定发送“短信1”了,A编辑如下短信内容:
我是A,我想进行提议,本次提议的时间戳是02:15
如前所述,我们暂时用时间戳代替序列号,并假设不会有两个人使用相同的时间戳。
A决定把“短信1”发给A,B,C三个人。A发给自己就不用真的发送了,他马上就知道了自己发送的内容。然后A,B,C三个人按照前面所述的规则对“短信1”做出回应。A刚刚发出“短信1”,E也决定发送“短信1”,E编辑如下短信内容:
我是E,我想进行提议,本次提议的时间戳是02:20
然后,E决定把他的"短信1"发送给C,D,E三个人。我们假设这些短信都在短时间内被接收到了。于是A,B,C收到了A的“短信1”,C,D,E收到了E的“短信1”,A和B马上给A做出回应:
我收到时间戳为02:15的提议请求,我同意你进行提议,并且,我之前没有同意过任何提议。
D和E也马上对E做出回应:
我收到时间戳为02:20的提议请求,我同意你进行提议,并且,我之前没有同意过任何提议。
C会收到A和E两个人的短信1,C具体怎么做,取决于收到短信的顺序。
情况1:C先收到了E的“短信1”或者C正在回复A的短信1但还没发送的时候收到了E的“短信1”,C都会决定不再理会A,只对E回复:
我收到时间戳为02:20的提议请求,我同意你进行提议,并且,我之前没有同意过任何提议。
在这种情况下,E会收到C,D,E三个人对“短信1”的回复,并开始发送“短信2”,而A不会收到3分对“短信1”的回复,于是等待一段时间后,A会发送一个时间戳更新的“短信1”。A可能只等待了一分钟,就迫不及待的发送了一个时间戳为02:16的“短信1”,如果C收到了的话,依然是不与理会,直到A发送的“短信1”的时间戳大于02:20,C才会理会A。

情况2:C先收到了A的“短信1”,并且,直到C回复A之前,都没有收到E的短信,C会回复A:
我收到时间戳为02:15的提议请求,我同意你进行提议,并且,我之前没有同意过任何提议。
然后,等C收到了E的“短信1”,C也会回复E:
我收到时间戳为02:20的提议请求,我同意你进行提议,并且,我之前没有同意过任何提议。
虽然C对A和E都进行了回复,但C回复的最新的时间戳是02:20,因此当C再收到A发来的时间戳为02:15的“短信2”的时候,C不会理会A的“短信2”。由于D和E回复了E的“短信1”,他们会记住他们接收的最新的时间戳是02:20,因此D和E收到A的时间戳为02:15的“短信2”时,也不会理会。C,D,E都不会理会A的“短信2”,所以A不可能收齐三个人对“短信2”的回复,因此,A的时间戳为02:15的提议,是不会得到通过的,A最终会由于很长时间都没有收齐3个对“短信2”的回复,而使用新的时间戳重新发送“短信1”

我们看,无论是情况1还是情况2,A的时间戳为02:15的提议都不会得到通过。而假如一切顺利的话,E收到3个对“短信1”的回复后,编辑“短信2”如下:
我提议把铅笔扔出窗外,我的时间戳是02:20。
然后,E决定把“短信2”发送给A,B,C。A和B在接收到E的“短信2”之前,接收到的最新的时间戳是A发出来的02:15,因此,A和B会发现他们收到的这条来自E的“短信2”的时间戳比他们以前收到的都新,按照前面约定好的规则,A和B会回复E:
同意你的时间戳为02:20的提议
并且,A和B都会记下:我最新同意的是时间戳为02:20的提议,提议的内容是仍铅笔。
对于C来说,他收到最大时间戳是02:20的“短信1”,这次他又收到了时间戳为02:20的“短信2”,按照前面规定的规则,出现并列最大的序列号时,需要对短信2做出回应,于是C回复E:
同意你的时间戳为02:20的提议
并且C会纪录下:我最新同意的是时间戳为02:20的提议,提议的内容是仍铅笔。

当A,B,C都决定同意E发出来的“短信2”之后,仍铅笔这个决定就已经被确定下来了。尽管A,B,C,D,E五个人还没有一个人知道这件事,但仍铅笔这件事已经确定了。比如,在A,B,C都回复C的“短信2”之后,B的回复由于通讯链路故障丢失了,E没有受到B的回复。这样,E就只收到了2个对“短信2”的回复。E会过一段时间重新发送“短信1”。

我们假设E在收到了两个A和C的两个“短信2”之后,还在等待B的回复。这时,C又发起了新一轮的提议,C编辑“短信1”如下:
我是C,我想进行提议,本次提议的时间戳是02:30
C决定把"短信1"发送给A,D,E,
按照规则,A会回复C:
我收到时间戳为02:30的提议请求,我同意你进行提议,并且,我之前同意过的最新的提议为"扔出铅笔",该提议的时间戳为02:20。
D和E都会回复E:
我收到时间戳为02:30的提议请求,我同意你进行提议,并且,我之前没有同意过任何提议。
C收到了A,D,E对“短信1”的回复之后,发现回复中提到的最新的提议是扔出铅笔,于是,C编辑“短信2”如下:
我提议把铅笔扔出窗外,我的时间戳是02:30。
C把“短信2”发送给任意3个人,比如A,B,D,假设运气好的话,他会收到3个这样的回复:
同意你的时间戳为02:30的提议
C自己知道他同意过的02:30的提议是扔出铅笔,于是C安心的把铅笔扔出窗外。为了加快达成一致的速度,C可以更新记录下:
我最新同意的是时间戳为02:30的提议,提议的内容是仍铅笔。
C也可以不更新上面这条记录,依然保持自己的记录为:
我最新同意的是时间戳为02:20的提议,提议的内容是仍铅笔。
回顾一下前面的流程,02:20这个记录内容是C回复E的短信2时记录下的。

C扔出去铅笔之后,就不再发送“短信1”或是“短信2”了,但会按照规则并依据自己收到的时间戳回复别人的“短信1”和“短信2”。最终,除去永远失联的0到2个人之外,其他人都会把铅笔扔出窗外。

接下来我们简单说说序列号的问题,paxos要求每个节点都能够生成全局唯一并且单调递增的序列号,当我一开始看paxos算法的时候,就在思考这个序列号如何生成。首先我想到的是时间戳,但两个不同节点上的时间戳可能是相同的,解决办法是给每个节点分配一个唯一的编号,比如上面的例子中,A,B,C,D,E五个人的编号分别为0,1,2,3,4,每个人想进行提案的时候,都获取一个时间戳,把这个时间戳转换为epoch(从1970年1月1日到现在的毫秒数),用这个epoch作为序列号的最高位,把自己的编号作为序列号的最低位。这样由于时间戳始终在不停的增长,序列号一定是单调递增的,当两个人在同一时间获取序列号时,由于每个人的编号不同,最低位不会相同,因此可以两个人的序列号不会一样。注意,我们要让epoch作为高位,每个人的编号作为低位,反过来是不行的。假如用每个人的编号作为高位,而epoch作为低位的话,一个人的序列号就永远无法超过另一个人的序列号了,而每个节点只要收到了一个大的序列号,就不会再处理小序列号的请求了,因此这会造成算法永远无法收敛。
进一步的,我们还可以得到一个更紧凑的序列号,假设一共有n个节点(在我们的例子中,n=5),每个节点的编号为i,i是从0到n-1的整数,epoch为时间戳。那么可以用如下公式得到序列号:
sn = epoch * n + i
比如n等于5的时候,如果按照一开始提出的序列号方案,sn号的低位只能用到了0,1,2,3,4,sn号永远不可能为5,6,7,8,9,这对于sn号的取值空间来说也是一种浪费,后面提出的这个公式可以让sn号的取值空间没有浪费。想到这里,我觉得我已经解决序列号的问题了,于是不再考虑生成序列号的问题。直到我无意中发现,google在Chubby的论文中提出了一个更简洁的方案:
系统中的每个节点都维护一个计数器,从0开始计数,每当这个节点打算新提出一个提议的时候,就把这个计数器加1,系统的总节点个数是n,每个节点都有一个唯一编号,从0到n-1,假设当前计数器的值为m,那么序列号为:
sn = m * n + i
对比我自己想的那个公式,把epoch换成了m,不需要用真实的时间了,只要每个节点维护一个单调递增的计数器就行。至此,序列号的问题算是完美的解决了。

说句题外话,raft算法所用的虚拟时钟相对于paxos lease算法的真实timer,看起来和chubby论文对我设想的序列号公式的简化非常神似。
基本的paxos算法 + paxos lease + multi paxos可以实现一个分布式的状态机系统。近些年提出来的raft算法号称可以实现paxos同样的功能,但更好理解。在我看来,raft本质和paxos是相通的,我倾向于认为raft就是进化到极致的paxos。

2014年11月13日星期四

pcaptraceroute,一个用来检测GFW的工具

从目前的经验来看,GFW对于IP在黑名单中的网站也并不是全部数据包都拦截,似乎是为了隐藏自己的存在,GFW只选择性的拦截返回包,而不拦截第一次发送出来的包。

现象1:你在境内的一台电脑上发送一个TCP syn包给境外一个被GFW拦截的服务器,在境外的服务器上用tcpdump抓包,会发现这个境内的电脑发送的syn包可以成功到达,同时,tcpdump也会显示,境外的服务器按照网络协议的要求返回了一个ack包给境内的电脑,但境内的电脑无法接受到这个返回的ack包。

现象2:从境外被GFW屏蔽的服务器上发送一个TCP syn包给境内的电脑,境内的电脑可以接受到这个syn包,但境内的电脑返回的ack包无法达到境外的服务器。

GFW这种方式让人很难检测出它的存在,因为你主动发过去的包是可以通过的,它只把返回的包拦截下来,这样在你看来是畅通的,但对方没有返回。因此,我写了一个pcaptraceroute程序,它用和traceroute同样的原理来检测网络中间节点的可达性,只不过,它发送的是TCP的ack包。这样,如果在一台被GFW屏蔽的主机上运行这个程序,就可以检测出数据包是从哪一跳开始完全不可达了。

项目主页:

2012年8月13日星期一

部分cfs调度器的tuning参数

最近看了rhel6.2内核的cfs进程调度代码,作为学习总结,把相关的部分tuning参数做了下整理,摘录如下。

* sched_child_runs_first
  当创建子进程时,保证子进程会在父进程之前运行。在创建子进程时,首先会
  将子进程的虚拟运行时间设置为min_vruntime,这个min_vruntime大约等于当
  前运行队列中所有进程的虚拟运行时间的最小值,然后,看sched_features中
  是否设置了START_DEBIT位,如果这一位被设置了,表示要给新创建的进程的
  虚拟运行时间再增加一些,这样会让它的运行时间向后延迟,以防进程通过不
  停的fork来不停的获得cpu时间。当设置完子进程的虚拟运行时间之后,会判
  断sched_child_runs_first是不是设置为1,如果是的话就比较父进程和子进
  程的虚拟运行时间,如果父进程的虚拟运行时间比较小的话,就交换父子进程
  的虚拟运行时间,这样子进程就会在父进程之前运行了。

* sched_min_granularity_ns
  这个值在两个地方会用到。一个是计算nice值为0的进程运行一次需要的时间。
  如果进程的总数大于sched_latency/sched_min_granularity个,就用
  sched_min_granularity_ns的值乘以进程的个数,将乘积作为nice值为0的进程
  调度一次运行的时间。然后每个进程会以此为基准,按照自己的优先级来计算
  自己被调度一次的运行时间。。另一个用途是判断当前进程是否要被调度下
  CPU的时候。如果当前进程的执行时间已经超过了它被调度一次所允许的执行时
  间(其实也可以说是时间片,尽管cfs声称自己没有时间片),那么必然要被调
  度下CPU。但如果它的时间片还没到期,在判断它的运行时间是不是比
  sched_min_granularity_ns小。如果是的话,那么就肯定不把它调度下CPU,如
  果进程的运行时间比sched_min_granularity_ns大的话,就用当前进程的虚拟
  运行时间减去runqueue上虚拟运行时间最小的进程的虚拟运行时间,如果差值
  比当前进程的时间片(这个时间片确实不是传统的时间片,总是随当前负载的
  状况变化)还大的话,那么当前进程也会被调度下CPU。

* sched_latency_ns
  计算nice值为0的进程运行一次所需的时间时用到。。如果进程总数小于等于
  sched_latency/sched_min_granularity个,就将sched_latency_ns作为nice
  值为0的进程运行一次的时间,否则按照前面介绍sched_min_granularity_ns
  时的方法计算。

* sched_wakeup_granularity_ns
  判断一个进程是否可以抢占当前进程时用到。如果该进程的虚拟运行时间比当
  前进程的虚拟运行时间大,那么肯定不能抢占。如果该进程的虚拟运行时间比
  当前进程的虚拟运行时间小的话,就计算出两者之差vdiff。如果vdiff大于一
  定的范围的话,就可以抢占,否则不可以抢占。
  sched_wakeup_granularity_ns就是用来确定这个范围的。它表示一个nice值
  为0的进程要抢占当前进程,它的虚拟运行时间需要被当前进程的虚拟运行时
  间小多少,其他优先级的进程以此为基准进行调整,优先级越高的进程,需要
  的差值越小。

* sched_tunable_scaling
  当内核试图调整sched_min_granularity,sched_latency和
  sched_wakeup_granularity这三个值的时候所使用的更新方法,0为不调整,1
  为按照cpu个数以2为底的对数值进行调整,2为按照cpu的个数进行线性比例的
  调整。

* sched_features
  每个bit都表示调度器的一个特性是否打开。在sched_features.h文件中记录
  了全部的特性。

* sched_migration_cost
  用来判断一个进程是不是cache hot的。如果进程的运行时间小于
  sched_migration_cost就认为是cache host的,在多CPU之间进行负载均衡的
  时候不会转移cache hot的进程。

* sched_nr_migrate
  在多CPU情况下进行负载均衡时,一次最多移动sched_nr_migrate个进程到另
  一个CPU上。

2012年8月4日星期六

关于linux内核抢占的一点思考

偶然看了一下rhel6.2系统内核中处理软中断的内核线程,很意外的产生了一点困
惑,这是ksoftirqd线程的主循环:

1    while (!kthread_should_stop()) {
2        preempt_disable();
3        if (!local_softirq_pending()) {
4            preempt_enable_no_resched();
5            schedule();
6            preempt_disable();
7        }
8
9        __set_current_state(TASK_RUNNING);
10
11        while (local_softirq_pending()) {
12            /* Preempt disable stops cpu going offline.
13               If already offline, we'll be on wrong CPU:
14               don't process */
15            if (cpu_is_offline((long)__bind_cpu))
16                goto wait_to_die;
17            do_softirq();
18            preempt_enable_no_resched();
19            cond_resched();
20            preempt_disable();
21            rcu_sched_qs((long)__bind_cpu);
22        }
23        preempt_enable();
24        set_current_state(TASK_INTERRUPTIBLE);
25    }

假设在23行执行完之后,有人试图唤醒这个线程,此时,线程还处于
TASK_RUNNING的状态,所以什么都不会发生。然后,在第24行,线程将自己设置
成了TASK_INTERRUPT状态,在执行第24行之后,线程被抢占了,那么,它在
TASK_INTERRUPT状态下被抢占,除非再次有人唤醒它,它岂不是再也返回不了了?
这样,不就丢失了一次唤醒吗?

这让我很困扰,于是,我自己写了个测试程序,启动一个内核线程,然后让它把
自己设置成TASK_INTERRUPTIBLE的状态,再进行几秒钟的忙循环。正常来说,执
行几秒钟的忙循环的话,肯定会被抢占了。然后我看它还能不能再被调度上来。
结果意外的发现rhel6的内核编译时是配置成不可抢占的(后来发现rhel5也这
样),我那个单核的虚拟机一执行忙循环就hang住。于是,重新编译了一个可抢
占的内核跑起来。结果发现,执行忙循环后,还是能被调度上来的。但如果我不
执行忙循环,而是调用schedule()函数,那么,除非被唤醒,否则就不能再被调
度上来了。

用systemtap跟踪了一下我的内核线程进入schedule()函数时的一些信息,结果
发现如果我主动调用schedule()函数,在进入schedule()函数时,thread_info
中的preempt_count的值是0x00000005,而如果在执行忙循环时被抢占调用
schedule函数时,preempt_count的值是0x10000005。这个最高位1在内核中被定
义为PREEMPT_ACTIVE。

原来,每当在抢占点上执行schedule函数时,这个标志会被设置,比如,从中断
返回时,调用:
retint_kernel
-> preempt_schedule_irq
   -> schedule

在preempt_schedule_irq函数中,就会调用这几行代码:

        add_preempt_count(PREEMPT_ACTIVE);
        local_irq_enable();
        schedule();
        local_irq_disable();
        sub_preempt_count(PREEMPT_ACTIVE);

设置PREEMPT_ACTIVE后,再调用schedule()函数,然后再清除PREEMPT_ACTIVE标
志。

而在schedule()函数中,有这样一段代码:

    if (prev->state && !(preempt_count() & PREEMPT_ACTIVE)) {
        if (unlikely(signal_pending_state(prev->state, prev)))
            prev->state = TASK_RUNNING;
        else
            deactivate_task(rq, prev, DEQUEUE_SLEEP);
        switch_count = &prev->nvcsw;
    }

prev就是当前要被调度下cpu的进程。prev->state为0表示TASK_RUNNING的状态。
如果进程的状态不是TASK_RUNNING并且PREEMPT_ACTIVE标志没有被设置,再判断进程是否有pending的信号,如果有,就把进程状态设置成TASK_RUNNING状态,
如果没有pending的信号,就把进程从RUNNING的进程队列里拿下来。

因此,逻辑关系是这样的:
如果PREEMPT_ACTIVE被设置了,说明进程是由于内核抢占被调度下CPU的,这时不
把它从RUNNING的队列里移除,如果进程不是由于内核抢占被调度下来的,看它
有没有未处理的信号,如果有的话,也不把它从RUNNING的队列里移除。只有再
上述两种情况都为假的情况下,进程才会被从RUNNING的队列里移除。

这就解释了为什么内核线程把自己设置成TASK_INTERRUPTIBLE状态之后,只要不
是主动调用schedule()函数,而是被抢占调度下cpu的,它就还会获得机会运行。

2012年7月27日星期五

linux内核的cfq IO调度算法的tuning参数

                              cfq_tuning
                              ==========

Author: yu peng
Date: 2012-07-27 19:58:01 HKT


Table of Contents
=================
1 概要
2 back_seek_max 和 back_seek_penalty
3 slice_async,slice_sync,和 slice_async_rq
4 low_latency
5 quantum
6 fifo_expire_async 和 fifo_expire_sync
7 slice_idle
8 group_idle
9 group_isolation


1 概要
-------
  以rhel6系统所用的2.6.32内核为基础,分析cfq io调度器的tuning参数。

2 back_seek_max 和 back_seek_penalty
-------------------------------------
  这两个参数是用来确定,对于给定的两个IO,哪个更适合作为“下一个”IO被
  传送给设备。判定的方法就是谁距离磁头更近。磁头的位置就是上一次IO完成
  的位置。比如磁头的位置last是sector 100。两个IO,io1的起始扇区号是105,
  io2的起始扇区号是110那么,io1距离磁头更近,下一次要发送IO,就先发io1,
  再发io2。
  如果io1的起始扇区号是90,io2的起始扇区号是110呢?两个io距离磁头的距离
  都是10,这时该如何选择呢?一般来说,把磁头向前移动(从扇区100向扇区
  90移动)需要花费的时间要多一些,所以cfq倾向于选择向后移动(从扇区100
  向扇区110移动)。
  如果io1的起始扇区号是95,io2的起始扇区号是110呢?虽然io1在磁头的前面,
  io2在磁头的后面,但io1距离磁头更近一些,这时该如何选择呢?具体的判断
  方法就用到了back_seek_penalty。比如io1在磁头的前面,距离磁头的距离是
  s1,io2在磁头的后面,距离磁头的距离是s2,如果s1*back_seek_penalty小于
  s2,就认为io1距离磁头比较近,否则,就认为io2距离磁头比较近。
  如果io1在磁头的前面,而io2在磁头的后面,但io1距离磁头太远了,超过了
  back_seek_max,那么无论io2是不是距离磁头更远,都选择io2。

3 slice_async,slice_sync,和 slice_async_rq
---------------------------------------------
  每个进程发下来的IO都被分成三类:异步,同步,idle。cfq先处理所有进程
  发下来的同步IO,然后处理所有进程发下来的异步IO,最后处理idle IO。进一
  步的,同步和异步的IO又被分别分成了8个不同等级的优先级。同一个进程发
  送下来的同步IO,都有同样的优先级,异步的也是如此。对于每个进程的IO,
  都被组织到一个叫做cfq_queue的结构体中。对于一个进程发送下来的IO,所
  有的同步IO有一个cfq_queue,所有的异步IO有另一个cfq_queue。cfq会给每
  个cfq_queue都分配一个时间片。比如当一个进程的同步IO的时间片用完后,就
  开始发送另一个进程的同步IO。时间片的大小由两个因素决定,一个是IO的优
  先级,这是和进程的IO优先级相关的,如果进程没有设置IO优先级,就按照
  nice值成正比计算出一个,另一个因素就是slice_async和slice_sync这两个
  参数。如果这两个参数越大,那么所有异步和同步IO的时间片粒度就越大,反
  之则粒度越小。对于异步IO,还有一个限制,如果在一个时间片内,已经发送
  的异步IO总数大于slice_async_rq,那么也视为时间片到期。idle类型的IO没
  有时间片,每次只发送一个。
  注意,这里说的异步IO和linux内核提供的AIO机制是两码事,一点关系都没有。
  一般来说,用户进程发送的读操作和设置direct标志的写操作都是sync类型的
  IO,普通的写操作会写道cache里面,再由内核进程通过cache发送到IO调度器
  的是async类型的IO。

4 low_latency
--------------
  如果这个参数被设置了,那么当前面计算出的cfq_queue的时间片大于
  low_latency时,会把时间片强行设置为low_latency。

5 quantum
----------
  除了idle类型的IO是一个一个的放到块设备的queue里面之外,sync和async类
  型的IO每次都是批量放入queue里面的。quantum就是一次批处理的数量。

6 fifo_expire_async 和 fifo_expire_sync
----------------------------------------
  cfq对不同进程发下来的IO分时间片进行处理,但在处理同一个进程发下来的
  IO时,采用与dead line类似的做法。尽量按照减小磁头移动的方式选择IO,
  但如果有些IO等待的时间太久了的话,就转而处理这些等待太久的IO。
  fifo_expire_async和fifo_expire_sync就是分别用来设置async类型IO和sync
  类型IO的超时时间的。

7 slice_idle
-------------
  在cfq中,最后被处理的IO是idle类型的IO,当调度器中已经没有任何的同步
  或异步类型的IO,并且这种状况已经持续了slice_idle这么长的时间,那么
  cfq就认为现在处于idle状态了,就发送一个idle类型的IO,然后再等待
  slice_idle这么长的时间,如果还是没有其他IO,就再发送一个idle类型的IO。
  补充一下,这个参数,以及所有本文中提到的和时间有关的参数,单位都是毫秒。

8 group_idle
-------------
  我的理解,不一定准确:
  当这个参数被设置之后,每当属于一个cgroup的进程的IO都提交完了,但时间
  片还没有耗尽,那么不会马上提交属于另一个cgroup的IO,而是会等待
  group_idle这么长的时间,如果在这个时间段内,该cgroup又有IO提交了进来,
  那么就继续处理。
  这个参数的引入是为了针对磁盘阵列或者固态硬盘这样的设备做优化用的,参
  看这篇文章:
  [http://lwn.net/Articles/395769/]
  以及内核文档blkio-controller.txt中的描述:
CFQ sysfs tunable
=================
/sys/block/<disk>/queue/iosched/slice_idle
------------------------------------------
On a faster hardware CFQ can be slow, especially with sequential workload.
This happens because CFQ idles on a single queue and single queue might not
drive deeper request queue depths to keep the storage busy. In such scenarios
one can try setting slice_idle=0 and that would switch CFQ to IOPS
(IO operations per second) mode on NCQ supporting hardware.

That means CFQ will not idle between cfq queues of a cfq group and hence be
able to driver higher queue depth and achieve better throughput. That also
means that cfq provides fairness among groups in terms of IOPS and not in
terms of disk time.

/sys/block/<disk>/queue/iosched/group_idle
------------------------------------------
If one disables idling on individual cfq queues and cfq service trees by
setting slice_idle=0, group_idle kicks in. That means CFQ will still idle
on the group in an attempt to provide fairness among groups.

By default group_idle is same as slice_idle and does not do anything if
slice_idle is enabled.

One can experience an overall throughput drop if you have created multiple
groups and put applications in that group which are not driving enough
IO to keep disk busy. In that case set group_idle=0, and CFQ will not idle
on individual groups and throughput should improve.

What works
==========
- Currently only sync IO queues are support. All the buffered writes are
  still system wide and not per group. Hence we will not see service
  differentiation between buffered writes between groups.

  在高速存储上,这样一组经验值可以获得较好的性能:
  slice_idle = 0
  quantum = 64
  group_idle = 1
  这些值在/sys/block/devicename/queue/iosched/中可以找到。

9 group_isolation
------------------
kernel document中有一个blkio-controller.txt文件,对这个参数做出了解释,
将其原文摘抄如下:
CFQ sysfs tunable
=================
/sys/block/<disk>/queue/iosched/group_isolation

If group_isolation=1, it provides stronger isolation between groups at the
expense of throughput. By default group_isolation is 1. In general that
means that if group_isolation=0, expect fairness for sequential workload
only. Set group_isolation=1 to see fairness for random IO workload also.

Generally CFQ will put random seeky workload in sync-noidle category. CFQ
will disable idling on these queues and it does a collective idling on group
of such queues. Generally these are slow moving queues and if there is a
sync-noidle service tree in each group, that group gets exclusive access to
disk for certain period. That means it will bring the throughput down if
group does not have enough IO to drive deeper queue depths and utilize disk
capacity to the fullest in the slice allocated to it. But the flip side is
that even a random reader should get better latencies and overall throughput
if there are lots of sequential readers/sync-idle workload running in the
system.

If group_isolation=0, then CFQ automatically moves all the random seeky queues
in the root group. That means there will be no service differentiation for
that kind of workload. This leads to better throughput as we do collective
idling on root sync-noidle tree.

我的理解,不一定准确:
cfq支持cgroup机制,可以区别对待属于不同cgroup的进程发下的io,给他们分
配不同的带宽。但这样会对效率造成比较大的影响。因为每个进程发送下来的读
写请求的位置可能差别很大,这样就会造成磁盘磁头来回的移动。所以,cfq提
供了这样一个参数,当group_isolation为0的时候,就把所有属于
SYNC_NOIDLE_WORKLOAD类型的cfq_queue都移动到root_group这个cgroup中。每
个进程都有几个它自己的cfq_queue,进程发送下来的IO,会根据IO的种类分别
放到不同的cfq_queue中去。从这个函数里我们可以看到怎么判断cfq_queue的种
类:
static enum wl_type_t cfqq_type(struct cfq_queue *cfqq)
{
        if (!cfq_cfqq_sync(cfqq))
                return ASYNC_WORKLOAD;
        if (!cfq_cfqq_idle_window(cfqq))
                return SYNC_NOIDLE_WORKLOAD;
        return SYNC_WORKLOAD;
}
如果cfq_queue里面放的是同步的IO,那么它就是ASYNC_WORKLOAD类的,如果
cfq_queue里面放的不是idle类型的IO(优先级最低的一种IO,只在空闲时才处
理),那么它就是SYNC_NOIDLE_WORKLOAD类型的,否则(也就是idle类型的IO),它
就是SYNC_WORKLOAD类型的。

2011年11月10日星期四

cfq io调度算法简介

本文以rhel5.6所使用的2.6.18内核进行分析。

1 cfq用到的几个数据结构
~~~~~~~~~~~~~~~~~~~~~~~~
  在cfq中,进程被按照优先级分组。cfq尽量优先处理高优先级组中的请求,同
  时也避免低优先级组中的请求被饿死。一个进程可能同时向多个不同的块设备
  提交io请求,而一个使用cfq调度算法的块设备又要处理多个进程中传下来的
  请求,并将它们分组。所以,cfq使用下面这些数据结构分别描述进程,优先
  级组,以及一个cfq块设备。

1.1 struct io_context
======================
   每个进程都有一个io_context结构体,从其定义中包含
   struct as_io_context *aic;
   和
   struct rb_root cic_root;
   这两位来看,它是专为as和cfq两个io调度算法准备的,在cfq中,它是
   struct cfq_io_context的基类。

1.2 struct cfq_io_context
==========================
   cfq_io_context是cfq中为每个进程分配的结构体,继承自io_context。每个
   进程都只有一个io_context,但它同时向多少个使用cfq调度算法的块设备提
   交io请求,就会有多少个cfq_io_context,这些个cfq_io_context被组织在
   一个红黑树中,树根是io_context中的cic_root,cfq_io_context中的"void
   *key"位用作红黑树的键值。由于每个cfq_io_context对应一个使用cfq调度
   算法的块设备,所以实际上cfq使用struct cfq_data结构体的地址作为键值,
   每个使用cfq算法的块设备都有唯一的一个struct cfq_data结构体。

1.3 struct cfq_queue
=====================
   当多个进程向同一个块设备提交io请求的时候,cfq会对进程按照优先级进行
   分组,进程的优先级信息存储在进程结构体的"unsigned short ioprio"位中,
   如果该位没有进行设置,cfq按照进程的nice值计算出一个优先级。默认共有
   8个优先级,对于每个优先级组,使用一个cfq_queue结构体进行表示。

1.4 struct cfq_data
====================
   每个使用cfq调度算法的块设备都有一个cfq_data结构,保存在struct
   elevator_queue的void *elevator_data位中。

1.5 struct cfq_rq
==================
   每个request都有一个cfq_rq,保存在request的void *elevator_private位
   中。

2 cfq_set_request函数
~~~~~~~~~~~~~~~~~~~~~~
  每当有一个新的io请求提交之后,电梯算法都会调用cfq_set_request函数让
  cfq为该请求分配必要的资源。
  首先调用cfq_get_io_context函数获取一个struct cfq_io_context结构,如
  果没有的话cfq_get_io_context函数会分配一个新的。
  接下来看cfq_io_context的"struct cfq_queue *cfqq"位有没有包含一个
  struct cfq_queue,如果没有的话就调用cfq_get_queue函数。
  cfq_get_queue函数计算当前进程的io优先级,并检察当前块设备的struct
  cfq_data结构中有没有对应该优先级的cfq_queue,有的话就返回,没有的话
  就分配一个新的。然后调用cic_set_cfqq函数让当前进程的cfq_io_context指
  向该cfq_queue。
  接下来分配一个新的对应于当前请求的struct cfq_rq结构。

3 cfq_insert_request函数
~~~~~~~~~~~~~~~~~~~~~~~~~
  调用完cfq_set_request函数后,电梯算法会调用cfq_insert_request将新提
  交的请求插入到cfq的数据结构中。
  首先检查cfq_queue的优先级是否改变,如果改变将相应的信息更新。然后将
  该请求对应的cfq_rq插入到以cfq_queue的"struct rb_root sort_list"为根
  的红黑树中。将请求按照提交时间的先后顺序放到cfq_queue的fifo链表上,
  如果该请求是可合并的,则将请求放到cfq_data的哈希表中,在合并请求时,
  会用到该哈希表。最后,调用cfq_crq_enqueued函数更新一些时间戳以及统计
  信息。

4 cfq_dispatch_requests函数
~~~~~~~~~~~~~~~~~~~~~~~~~~~~
  每当电梯算法要补充一些请求到请求队列上的时候,会调用
  cfq_dispatch_request函数。
  首先调用cfq_select_queue选择一个优先级队列,然后调
  用__cfq_dispatch_requests函数将该优先级队列上的一部分请求放到请求队
  列上。
  选择优先级队列时,总是从最高等级的优先级队列开始选择,但每个等级都有
  一个时间片,如果该等级的时间片到期,则选择下一个等级的队列。

5 merge
~~~~~~~~

5.1 cfq_merge函数
==================
   电梯算法调用cfq_merge函数判断一个新的请求是否能和以前的请求合并。
   首先判断是否能进行后向合并。cfq_data中有一个哈希表,储存了所有可以
   合并的请求,并且以请求的结束扇区号作为键值。所以,以当前请求的起始
   扇区号作为键值进行搜索,如果能够搜到一个请求,则说明搜到的请求的结
   束扇区号和当前请求的起始扇区号相等,可以进行后向合并。
   如果没有找到能够进行后向合并的请求,则尝试寻找是否有能进行前向合并
   的请求。以当前请求的结束扇区号为键值,在与当前请求的io优先级相等的
   cfq_queue的红黑树中寻找。由于该红黑树中的成员是以请求的起始扇区号为
   键值的,如果找到,说明可以进行前向合并。
   cfq_merge的返回值会告诉调用它的函数是可以进行前向合并,还是后向合并,
   还算不可以合并。如果可以合并,会将可用来合并的请求放到struct
   request **req中返回。

5.2 cfq_merged_request函数
===========================
   所谓合并,就是将一个新传递进来的bio放到一个以前的request中。实际的
   合并操作是在ll_rw_blc.c和elevator.c中完成的。在合并完成之后,电梯算
   法会调用cfq_merged_request函数让cfq进行一些必要的更新。由于进行了合
   并,request的起始扇区号和结束扇区号可能会发生变化,而cfq内部所用的
   哈希表以及红黑树是以这两个值作为键值的。所以,cfq_merged_request的作
   用就是检验这两个值是否变化,若变化了则以新的键值放置相应的数据结构
   成员

2011年10月18日星期二

linux的deadline io 调度算法分析

1 linux的deadline io 调度算法分析
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
  以rhel5.6系统(kernel version 2.6.18)为基础分析linux内核的块设备层
  的deadline io 调度算法。

2 把bio添加到请求中以及把请求添加到请求队列中
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

2.1 最简单的情况,请求队列中一个请求都没有的时候,添加第一个bio
================================================================
   bio通过generic_make_request函数传递进来。generic_make_request函数会
   调用请求队列的make_request_fn函数指针完成实际的io操作。如果使用电梯
   算法,请求队列的make_request_fn函数指针将指向__make_request函数。
   我们只考虑与电梯算法相关的部分。

   首先,调用elv_merge函数检查是否可以将当前的bio合并到以前的request
   中。如果是第一次执行到这里,当然没有可以合并的request,于是调用
   get_request_wait函数获取一个新的request,然后调用
   init_request_from_bio函数将bio添加到request中。接下来调用
   add_request函数将request添加到request_queue中。

   add_request中的函数调用流程:
   add_request->__elv_add_request->elv_insert
   elv_insert函数会将新创建的request挂到request_queue的queue_head链表
   上,并让request_queue的last_merge指针指向当前的request。最
   后,elv_insert函数会调用到函数指针:
   q->elevator->ops->elevator_add_req_fn
   如果deadline是该块设备的io调度算法的话,这个指针就指向了
   deadline_add_request函数。

   deadline算法中有两个红黑树(读写请求各有一个),两个双向链表(读写
   请求各有一个),一个哈希表(所有读写请求都在同一个哈希表上)。

   deadline_add_request函数会将传递进来的request添加到与请求的读写方
   向相对应的红黑树与双向链表中,以及哈希表上。同时,也记录下这个请求
   的到期时间。

   deadline中的红黑树以起始扇区号作为键值进行排序。双向链表按照request提
   交进来的先后顺序进行排序,最先加进来的request放在链表的头部。因此,每
   新添加一个请求的时候,都放在链表的尾部。哈希表使用request的结束扇
   区号(即rq->sector + rq->nr_sectors)作为键值进行索引。读请求的期
   满值默认为0.5秒,写请求的期满值默认为5秒。在合并与分发请求时,我们
   会看到deadline算法如何使用这些数据。

   调用完add_request函数后,__make_request的任务基本就结束了。每一个
   请求队列,在真正处理请求的时候都会有一个延迟,一般默认设为3毫秒。
   如果这3毫秒内又有request到来,则将该请求提交给io调度算法,并重新设
   置延迟的期满时间为3毫秒。

2.2 将bio merge到request中
===========================
   每当一个bio提交到__make_request函数中的时候,该函数总是调用
   elv_merge函数检查是否能将当前的bio合并到以前的request中。elv_merge
   函数分两步进行判断。

   第一步,判断判断bio是否可以合并到request_queue的last_merge指针指向
   的request。last_merge总是指向上一次提交进来的请求。如果进行连续读
   写的话,这是可能性最大的情况。判断的方法很简单,首先看request的结
   尾扇区号是否等于bio的起始扇区号,如果是的话就能够进行back merge,
   然后判断request的起始扇区号是否等于bio的结尾扇区号,如果是的话就能
   够进行front merge。

   第二步,如果bio不能和last_merge指向的request进行合并的话,则调用io
   调度算法的elevator_merge_fn成员函数,判断bio是否能和某个request进
   行合并。

   对于deadline调度算法,elevator_merge_fn指向deadline_merge函数。该
   函数首先判断是否能进行back merge。以bio的起始扇区号为键值在哈希表
   中寻找。由于deadline中的哈希表是以request的结束扇区号为键值的,所
   以如果找到这样的request的话,说明这个request的结语扇区号与bio的起
   始扇区号相等,因此bio可以back merge到这个request中。如果没有找到这
   样的request,说明无法进行back merge,则继续尝试能否进行front merge。
   方法是以bio的结束扇区号为键值在红黑树中进行搜索。红黑树是以request
   的起始扇区号为键值进行排序的,因此,如果找到符合条件的request,说
   明该request的起始扇区号等于bio的结束扇区号,可以进行front merge。
   无论是back merge还是front merge,如果找到符合条件的request的话,都
   要把request从哈希表上拿下来,再按照新的键值重新放上去,至于红黑树,会
   在后续的操作中进行处理。

   若是可以进行合并,elv_merge会返回用来合并的request指针,并且,通过
   返回值表示是何种合并,ELEVATOR_BACK_MERGE表示back merge,
   ELEVATOR_FRONT_MERGE表示front merge,ELEVATOR_NO_MERGE表示无法进行
   合并。

2.2.1 back merge
-----------------
    若是进行back merge,则调用函数指针q->back_merge_fn对request_queue
    和request进行一些必要的处理,这个指针通常指向ll_back_merge_fn函数。
    然后修改request的biotail,biotail->next,以及nr_sector等位,将bio
    添加到request中。接下来调用attempt_back_merge函数,该函数的作用是
    检查是否可以进一步的进行合并,例如,原先请求队列中有这样两个请
    求,request1写块设备的secotr0到sector4,request2写块设备的
    sector 8到sector 12。现在新加入的bio写sector5到sector7,bio被合并
    到了request1中,request1变成写sector0到sector7,此时,request1和
    request2可以进行合并了。若发生这种情况,attempt_back_merge函数会
    进行处理,若没有,则调用elv_merged_request函数进行最后的收尾处理。
    该函数会调用调度器的elevator_merged_fn方法,对于deadline调度器,
    该方法指向deadline_merged_request函数。该函数的作用很简单,由于进
    行了合并,request的起始扇区号和结束扇区号可能改变了,用新的扇区号
    做键值更新request在哈希表和红黑树中的位置。

2.2.2 front merge
------------------
    与back merge的处理基本对称,不再赘述。

3 从请求队列中取出一个请求
~~~~~~~~~~~~~~~~~~~~~~~~~~~
  正常情况下,每当request_queue等待3毫秒还没有新的bio提交后,就会出发
  unplug,最终会调到request_queue的request_fn方法。以scsi磁盘驱动为
  例,该方法指向scsi_request_fn函数。scsi_request_fn会调用
  elv_next_request函数来从请求队列中取出一个请求进行处理。而该函数会
  调用__elv_next_request找出一个合适的请求。

  __elv_next_request函数首先从request_queue的queue_head链表上寻找
  request,如果找到的话就返回这个request,没找到的话就调用调度器的
  elevator_dispatch_fn方法,通常,这个方法会将一些request放到
  queue_head链表上。然后再检查request_queue的queue_head链表。一直循
  环,直到从queue_head上找到一个request,将其返回,或者
  elevator_dispatch_fn方法返回0,则__elv_next_request函数返回一个空指
  针。

  对于deadline调度器来说,elevator_dispatch_fn函数指针指向
  deadline_dispatch_requests函数。

  deadline_dispatch_requests函数负责从已经提交给调度器的request中选择
  一个放到request_queue的queue_head链表中。

  为了优化读写磁盘的效率,deadline调度器会对请求进行批处理,每当选定
  一个请求方向时(读或写),总是连续处理该方向上的16个请求。然后才重
  新选择另一个方向上的请求。当然,如果某个方向上已经没有请求了,则马
  上选择另一个方向上的请求。

  每当处理了一个方向上的请求,就将该方向上的下一个请求放到
  next_drq[WRITE]中或next_drq[READ]中。因此,当进入
  deadline_dispatch_requests函数的时候,首先判断next_drq[WRITE]或
  next_drq[READ]是否为空,这两个指针至多有一个不为空。如果其中有一个
  不为空,我们就获得了一个request。接下来判断这个request和上一次发送
  的请求是否是连续的,若不是连续的,则打破批处理,然后判断当前的批处
  理进行了多少次,如果超过了最大的允许次数,也要打破批处理,重新选择
  request。这是为了防止在某一个方向上有大量连续的request,导致饿死了
  另一个方向上的request。

  如果确定要继续批处理,则直接跳到函数的末尾,即标签
  “dispatch_find_request”处,递增批处理的统计次数,将同方向上的下一个
  请求放到next_drq[WRITE]或next_drq[READ]中,记录当前request的结束扇区
  号,以便再进入本函数时判断下一个request是否和当前这个是连续的。最后
  把这个请求从deadline算法内部维护的哈希表,红黑数和链表上面移除,放到
  request_queue的queue_head链表上。

  如果前面判断不进行批处理,则先看是否有读请求需要处理。若读写请求都
  有的话,就把starved变量加一,表示忽略了一次写请求。如果忽略的写请求
  次数超过了变量writes_starved的值(默认为2),那么就不再忽略写请求,
  跳转到标签“dispatch_writes”的位置,将变量data_dir设为WRITE,表示提
  交一个写请求。否则将data_dir设为READ,表示提交一个读请求。

  现在已经选择了request的方向(READ或WRITE),接下来判断在该方向上最
  先进来的request是否已经超时了,或者该方向上还有没有request的起始扇
  区号比上一个请求更靠后的request了(这样的request即使不连续也可以让
  磁盘的磁头大体上朝一个方向移动)。如果两个条件中有一个条件为真,就
  选择该方向上时间最早的request提交。如果这两个条件都为假的话,即没有
  任何请求超时,而且有一个起始扇区号比上一个request大的请求,就提交这
  个请求。

  完。

2011年9月5日星期一

linux内核中SH_DIV宏分析

在读linux内核中关于时钟部分的代码时,遇到了一个叫SH_DIV的宏,它包含的
技巧颇为有趣。自己写代码时可能用得到。

下面是这个宏的注释和定义:

/* Suppose we want to devide two numbers NOM and DEN: NOM/DEN, the we can
 * improve accuracy by shifting LSH bits, hence calculating:
 *     (NOM << LSH) / DEN
 * This however means trouble for large NOM, because (NOM << LSH) may no
 * longer fit in 32 bits. The following way of calculating this gives us
 * some slack, under the following conditions:
 *   - (NOM / DEN) fits in (32 - LSH) bits.
 *   - (NOM % DEN) fits in (32 - LSH) bits.
 */
#define SH_DIV(NOM,DEN,LSH) (   (((NOM) / (DEN)) << (LSH))              \
                             + ((((NOM) % (DEN)) << (LSH)) + (DEN) / 2) / (DEN))


计算 (N << L) / D,可以等价于下面的算式:
(N / D) << L + ( (N % D) << L + (D / 2) ) / D

下面是推导过程:

N = (N / D) * D + (N % D)

N << L = N * 2^L
       = ( (N / D) * D + (N % D) ) * 2^L
       = (N / D) * D * 2^L + (N % D) * 2^L

(N << L) / D = ( (N / D) * D * 2^L + (N % D) * 2^L ) / D
             = ( (N / D) * D * 2^L ) / D + ( (N % D) * 2^L ) / D
         = (N / D) * 2^L + ( (N % D) * 2^L ) / D
         = (N / D) << L + ( (N % D) << L ) / D

这时与最终的等式只差一个(D / 2),当在C语言中进行除法运算(N / D)时,结
果总是向下取整,余数部分全都被舍去了,因此,在计算除法前给被除数加上除
数的一半,可以起到四舍五入的效果,即:(N + (D / 2)) / D。
所以,在前面推导过程的最后一步时,在除以D前要加上(D / 2),即:
(N << L) / D = (N / D) << L + ( (N % D) << L + (D / 2) ) / D

2011年5月9日星期一

甄姬洛神后摸桃子的概率

我考虑的是这样一种情况:

不包括任何扩展牌,牌堆中共104张牌,52红,52黑,红牌中有8张桃子。牌洗好
后甄姬开始洛神,直到洛神失败,然后抓两张牌。计算这种情况下甄姬至少摸到
一张桃子的概率。

首先,看白板武将至少摸到一张桃子的概率如何计算 :

先 计算白板武将一张桃子都摸不到的概率,然后用1减去它得到的就是至少摸到
一张桃子的概率。白板武将摸两张牌,相当于从104张牌中随机抽出两张牌,共有
104*103种抽法。不是桃子的牌共有96张,抽出两张不是桃子的牌,共有96*95种
抽法。所以,白板武将摸不到桃子的概率p1 = (96*95)/(104*103)。1-p1得到的
就是白板武将至少摸到一张桃子的概率。

为了给计算甄姬摸桃子的概率做铺垫,再考虑另一种计算白板武将摸桃子的概率。
还是先考虑一张桃子的摸不到的概率。

104 张牌进行排列组合,共有104的阶乘,即104!种排列方法。再看看前两张牌都
不是桃子共有多少种排列方法:先从96张不是桃子的牌中取一张放在最上面, 再
从剩下的95张不是桃子的牌中取1张放在牌库顶的第二张,再将剩下的102张牌任
意排列,共有:96*95*102!种排列方法。因此,白板武将摸不到 桃子的概率p1
= (96*95*102!)/(104!) = (96*95)/(104*103)与前一种方法的计算结果一致。

现在,我们来看甄姬洛神直到失败后,至少摸到一张桃子的概率。依然,我们还
是计算甄姬一张桃子都摸不到的概率,然后用1减去它得到甄姬至少摸到一张桃子
的概率。

104张牌进行排列组合,共有104!种排列方法。

甄姬洛神后抓牌,共有如下这么多种可能:

0 洛神到0张黑牌,即一张牌都没洛到。0.1 洛神的判定牌不是桃子从 44张不是
桃子的牌中任选一张放在牌库顶,不是桃子的牌共96张,其中有一张红的作为判
定牌放在牌库顶了,从剩下的95张中选一张放在牌库顶的第二张的位 置,再从剩
下的94张中选一张放在第三张的位置(我们现在计算的是1张桃子都摸不到的概率,
所以从不是桃子的牌中选2张给甄姬摸),1张放在了牌库顶,不 是桃子的牌中选
出了两张给甄姬摸,总共104张牌,还剩101张牌可以任意排列,共有101!种排列
方法。因此,洛神第一张就是红色,并且不是桃子共有这末多种情况:44 *
95 * 94 * 101!  0.2 洛神的判定牌是桃子与0.1的算法类似,从8张桃子中选一
张放在牌库顶,然后从96张不是桃子的牌中选1张放在牌库顶第二张,再从剩下的
95张不是桃子的牌中选一张放在牌库顶第三张,剩下的101张牌任意排列,共有这
末多种情况:8 * 96 * 95 * 101!

1 洛神到1张黑牌。从52张黑牌中选1张放到牌库顶,共有52种情况,接下来:
1.1 判定牌不是桃子从44张不是桃子的红牌中选1张放到洛神的牌后面,洛神1张
牌加上判定牌1张,还剩94张不是桃子的牌,从中选一张放在判定牌后面,再从剩
下的93张不是桃子的牌中选一张放在后面,此时包括桃子总共还剩100张牌,可以
任意排列,于是共有这末多种情况:52 * 44 * 94 * 93 * 100!  1.2 判定牌是
桃子从8张桃子中选1张放到洛神的牌后面,除去洛神的1张牌,还有95张牌不是桃
子,从中选1张放到判定牌后面,再从剩下的94张不是桃子的牌中选1张放到后面,
然后剩下的100张牌任意排列,共有这末多种情况:52 * 8 * 95 * 94 * 100!

2 洛神到2张黑牌从52张黑牌中选1张放到牌库顶,再从剩下的51张黑牌中选1张放
到牌库顶第二张,共有52*51种情况2.1 判定牌不是桃子从44张不是桃子的红牌中
选1张放到洛神的牌后面,洛神2张加上判定牌1张,还剩93张不是桃子的牌,从中
选1张放到判定牌后面,再从剩下的92张不是桃子的牌中选1张放到后面,一共还
剩99张牌,可以任意排列,共有这末多种情况:(52 * 51) * 44 * 93 * 92 *
99!  2.2 判定牌是桃子从8张桃子中选1张放到洛神的牌后面,除去洛神2张牌,
还剩94张不是桃子的牌,从中选1张放到判定牌后面,再从剩下的93张不是桃子的
牌中选1张放到后面,一共还剩99张牌,可以任意排列,共有这末多种情况:
(52 * 51) * 8 * 94 * 93 * 99!

依此类推,可以得到:

3 洛神到3张黑牌从52张牌中选3张放到牌库顶,共有52 * 51 * 50种情况3.1 判
定牌不是桃子(52 * 51 * 50) * 44 * 92 * 91 * 98!  3.2 判定牌是桃子

(52 * 51 * 50) * 8 * 93 * 92 * 98!

......

51 洛神到51张黑牌从52张牌中选51张放到牌库顶,共有52 * 51 * 50 *
...... * 2种情况51.1 判定牌不是桃子(52 * 51 * ... * 2) * 44 * 44 *
43 * 50!  51.2 判定牌是桃子(52 * 51 * ... * 2) * 8 * 45 * 44 * 50!

52 洛神到52张黑牌52.1 判定牌是桃子(52 * 51 * ... * 2 * 1) * 44 * 43 *
42 * 49!  52.2 判定牌不是桃子(52 * 51 * ... * 2 * 1) * 8 * 44 * 43 *
49!

将上面的所有情况相加,得到的就是所有洛神后摸不到桃子的情况。经计算得到,
洛神后摸不到桃子共有这末多种情况:

no_peach =
8768393644112035467653013895319007081912845854144094315049335792122609241987921866302232452367163796631068635340901979516433835590933660303360000000000000000000000000

又由于104张牌全排列共有104!种情况

因此,甄姬摸不到桃子的概率p2 = no_peach / 104!

no_peach刚好可以被102!整除,因此计算p2时分子分母同时约去102! 得到:p2
= 9120 / (104 * 103) = (96 * 95) / (104 * 103) = p1

即,甄姬洛神后摸不到桃子的概率,和白板武将摸两张牌后摸不到桃子的概率是
一样的,所以,他们至少摸到1张桃子的概率也是一样的,均为1 - (96 * 95) /
(104 * 103),约等于0.148618371919。

我把计算no_peach的值的程序放到github上了

https://github.com/yupeng820921/luoshen

calc.py

是个python写的小程序。
   

2011年2月11日星期五

linux内核raid5一次整条带写操作的流程(基于2.6.33内核)

Table of Contents
=================
1 raid5写入数据的顺序
2 make_request函数
        2.1 rand5_compute_sector
        2.2 get_active_stripe
            2.2.1 get_free_stripe
            2.2.2 init_stripe
                2.2.2.1 raid5_build_block
            2.2.3 get_active_stripe的其他部分
            2.2.4 __find_stripe
        2.3 add_stripe_bio
        2.4 在add_stripe_bio和release_stripe之间的操作
        2.5 release_stripe
            2.5.1 __release_stripe
3 raid5d
    3.1 __get_priority_stripe
    3.2 handle_stripe
        3.2.1 handle_stripe5
            3.2.1.1 handle_stripe_dirtying5
                3.2.1.1.1 schedule_reconstruction
            3.2.1.2 raid_run_ops
            3.2.1.3 handle_stripe5结束
    3.3 release_stripe
4 ops_complete_reconstruct
5 再次进入raid5d
    5.1 ops_run_io
6 raid5_end_write_request
7 第三次进入raid5d
    7.1 handle_stripe_clean_event
    7.2 release_stripe


1 raid5写入数据的顺序
######################
  假设,stripe size是4k大小,chunk size是32k,共有4块盘。首先,第一块
  盘的4k数据会写入第一块盘的第一个chunk的第一个stripe,如图1所示:



图1






图中桔黄色的方块表示写入的数据,蓝色的方块表示该处将用于存储校验数
  据。图中只画出了第一个chunk,所以校验数据都存于最后一个磁盘中。
  接下来的4k数据会写入第一块盘的第一个chunk的第二个stripe,如图2所示:




图2

直到写入8次4k的数据后,第一块盘的第一个chunk全部写完,如图3所示:


图3
写完这32k数据后,接下来的4k数据会写入第二块盘的第一个chunk的第一个
  stripe,如图4所示:



图4

依此类推,直到第三块盘的第一个4k数据写完,此时,第一个stripe即
  stripe0就被填满了,如图5所示:



图5
在raid5的make_request函数中,如果能够找到一个stripe来描述当前这次写
  操作,则将该stripe提交到一个全局链表中,然后唤醒守护进程raid5d来进
  行处理,make_request并不等到raid5d将数据真正写入磁盘,就直接返回了。
  所以,有大量的数据连续写入时,make_request函数会被连续调用,每次使
  用一个stripe来描述这次任务,就返回了。默认一共有256个stripe可用。当
  这256个stripe都使用完了之后,再有写操作的时候,make_request函数就会
  进入休眠,直到至少有四分之一的stripe被释放之后才会被唤醒,使用新释
  放出来的stripe来描述新的写操作。
  我们来看一下第一个stripe(即stripe0)中的数据的处理流程。

2 make_request函数
###################
  对一次正常的写操作,make_request函数中需要调用4个比较重要的函数。下
  面依次介绍。

2.1 rand5_compute_sector
~~~~~~~~~~~~~~~~~~~~~~~~~
    首先会调用raid5_compute_sector函数通过整个md设备的扇区号
    logical_sector换算出在实际磁盘上对应的扇区号new_sector,以及需要使
    用的是第几块磁盘dd_idx。我们只看和stripe0相关的部分。
    在本例中,第一次写入磁盘的4k数据会落入到stripe0中,即
    logical_sector=0,此时计算出的new_sector=0,dd_idx=0,即写入第一块
    磁盘的第一个sector。
    然后,在写入第九个4k的数据的时候,又落入到stripe0中,此
    时,new_sector = 0, dd_idx = 1, logical_sector = 64。
    在写入第17个4k的数据时,会再次落入到stripe0中,此时,new_sector =
    0, dd_idx = 2, logical_sector = 128。
    至此,第一个stripe被填满。
    每次调用raid5_compute_sector函数后,会返回new_sector和dd_idx两个变
    量。同一个stripe中的每块盘的扇区号是一样的,所以通过new_sector的值
    来判断本次写操作发生在哪个stripe中,并使用一个stripe_head类型的结
    构体来描述这次写操作。

2.2 get_active_stripe
~~~~~~~~~~~~~~~~~~~~~~
    分配stripe_head的操作是在get_active_stripe函数中完成的。首先,这个
    函数会尝试调用__find_stripe函数查看本次操作所在的stripe是不是已经
    在使用中了,如果是的话,就不用新分配stripe了。
    在第一次写入4k数据时,stripe0肯定不在使用中,所以__find_stripe函数
    的返回值是NULL,然后,会调用get_free_stripe函数去获得一个空闲的
    stripe。

2.2.1 get_free_stripe
======================
     进入get_free_stripe函数看一下,该函数很简单,从inactive_list链表
     中取下一个sh,然后给active_stripes的值加一,表示又有一个stripe被
     使用了。此外,还会将sh从它所在的hash表中拿下来(如果它确实挂在一
     个hash表上的话)。这个hash表是在__find_stripe中使用的,介绍后续的
     写操作时会进行说明。

2.2.2 init_stripe
==================
     如果成功的分配到了一个sh,就会调用init_stripe函数对该sh进行初始化。
     包括这个stripe的起始扇区号,使用第几块盘存放校验数据,以及将它的
     状态sh->state设成0。后面我们会看到这个状态在不停的变化。
     对于属于这个sripe的每一个块设备,都有一个r5dev类型的结构体来描述。
     这些个r5dev型的结构体也需要初始化。首先将表示该块设备状态的标志
     dev->flags清零。然后调用raid5_build_block函数初始化与bio相关的变
     量。

2.2.2.1 raid5_build_block
--------------------------
      raid5_build_block函数首先设置提交一次bio所使用的bi_io_vec以及存
      放数据的page之类的变量。然后会调用这样一个函数:
      dev->sector = compute_blocknr(sh, i, previous);
      这里计算出的sector是该磁盘在该stripe中的起始扇区,在整个md设备中
      的扇区号。

2.2.3 get_active_stripe的其他部分
==================================
     在get_active_stripe函数的末尾部分,有这样一句话:
     atomic_inc(&sh->count);
     将sh的记数器加1,后面会介绍到,每当进入release_stripe函数时,会将
     count的值减1,如果减到0了就表示当前对这个sh的处理都完成了,唤醒守
     护进程,来决定对sh的下一步处理。

2.2.4 __find_stripe
====================
     对于写入raid设备的第一个4k的数据,__find_stripe函数会返回空。但当
     写入第九个4k的数据时,也就是像图2中所示的情况发生时,__find_stripe函
     数通过sector为key值在一个全局hash表中查找,由于这时的sector值和第
     一个4k数据的sector值是一样的,而写入第一个4k数据时,在init_stripe
     函数中已经把那时分配到的sh以sector为key值挂到hash表中了,所以这
     时__find_stripe函数会找到写入第一个4k数据所用的sh,也就是用来描述
     stripe0的sh。

2.3 add_stripe_bio
~~~~~~~~~~~~~~~~~~~
    add_stripe_bio函数写在一个条件表达式中的不起眼的位置上,但功能很重
    要。
    首先判断是要执行读操作还是写操作。
    然后用一个临时变量bip指向写磁盘时需要使用的bio(即towrite)的地址,
    后面操作bip就等于改变了towrite的值。然后判断是否发生了overlap的情
    况,也就是前面已经有一个读写请求发生在这个stripe的这块盘上,而本次
    操作又发生在同一个stripe的同一块盘上,并且两次读写数据的位置还有重
    叠。对于我们的顺序写操作,肯定是不会发生这种情况的。
    接下来,对于写操作,要判断是否发生了overwrite的情况。所谓
    overwrite,就是这次写操作是不是覆盖了stripe在这块磁盘上的整个区间。
    如果是的话,在计算xor校验值的时候,对于这块磁盘,就直接使用上层传
    下来的数据,如果不是的话,就需要读回stripe在这块磁盘上的数据到内
    存,然后在把上层传下来的数据覆盖到内存,再计算xor,最后把内存中的
    数据写入磁盘。
    判断写入数据是否覆盖整个stripe的方法也很简单,如果写入数据的起始扇
    区号小于等于stripe的扇区号(bi->bi_sector <= sector),并且写入数据的
    结束扇区号大于等于stripe的结束扇区号(sector >=
    sh->dev[dd_idx].sector + STRIPE_SECTORS),那么这次写操作就是
    overwrite的。
    在我们的例子中,每次写入4k数据,刚好等于stripe的大小,所以是
    overwrite的。因此,下面语句会被调用:
    set_bit(R5_OVERWRITE, &sh->dev[dd_idx].flags);

2.4 在add_stripe_bio和release_stripe之间的操作
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
    此时make_request函数中会执行两步比较有意义的操作:
    set_bit(STRIPE_HANDLE, &sh->state);
    clear_bit(STRIPE_DELAYED, &sh->state);
    设置STRIPE_HANDLE位很重要,它表示这个stripe需要进一步的处理,在
    release_stripe函数中会通过该位来判断是否需要唤醒守护进程。

2.5 release_stripe
~~~~~~~~~~~~~~~~~~~
    接下来进入make_request中会调用到的最后一个比较重要的函数。
    release_stripe函数只有几行:获取锁,调用__release_stripe函数,释放
    锁。所以真正重要的操作都在__release_stripe函数中。

2.5.1 __release_stripe
=======================
     首先会判断sh的引用计数sh->count,如果它减小到0则说明所有对sh的操
     作都完成了,需要唤醒守护进程进行下一步的处理。
     我们记得,在前面调用过的init_stripe函数中有这样一行语句:
     BUG_ON(atomic_read(&sh->count) != 0);
     就是说一开始获取到的sh,引用计数肯定是0,然后,在
     get_active_stripe函数的的末尾处,会执行这样一行语句:
     atomic_inc(&sh->count);
     将sh的引用计数加1。
     对于stripe0共进行了3次4k数据的读写,每次进入get_active_stripe函数
     后,会让sh->count加1,进入__release_stripe函数后,又将其减到0。
     然后检测sh->state,目前sh->state的值应该是0x00000004,只有
     STRIPE_HANDLE被置位(在make_request函数调用add_stripe_bio之后,调
     用release_stripe函数之前)。因此在条件判断中会进入这一行:
     list_add_tail(&sh->lru, &conf->handle_list);
     将stripe挂在handle_list链表上。
     然后执行:
     md_wakeup_thread(conf->mddev->thread);
     唤醒raid5d守护线程。
     虽然守护线程会在这里被唤醒,但并不会马上执行。实际上,除非
     make_request函数由于sh耗尽而进入休眠,否则在它返回之前raid5d线程
     是一直都得不到机会执行的。所以,当写入stripe0的3块4k数据都执行
     过__release_stripe之后,再等到make_request函数返回或进入休
     眠,raid5d线程会开始执行,提交到strpe0的数据会得到进一步处理。

3 raid5d
#########
  raid5d线程开始执行后,会进入一个大循环,这个循环中,首先调
  用__get_priority_stripe函数获取一个需要处理的sh,然后调用
  handle_stripe函数对这个sh进行处理,最后调用release_stripe函数来决定
  是否要再次唤醒raid5d线程,或是把sh释放掉。一直等到所有的sh都处理完
  了,才退出循环。在raid5d的最后会调用async_tx_issue_pending_all函数,
  如果在处理过程中有进行dma操作的话,这个函数会确保dma开始运行。

3.1 __get_priority_stripe
^^^^^^^^^^^^^^^^^^^^^^^^^^
   这个函数会尝试从handle_list和hold_list两个链表上获取sh,对于写操作
   来说,如果一个sh已经是整条带写,那么它会被挂在handle_list上,否则就
   会挂在hold_list上。所以总是优先搜索handle_list,如果有整条带写,就
   先处理。等到整条带写的都处理完了,可能已经又有新的写操作提交进来,
   让那些之前不是整条带写的sh也变成整条带写了。这样可以提高性能。
   在这里,将stripe0的sh从handle_list上取下来,然后给sh->count加1。通
   常,把sh放到链表上的操作都是在release_stripe函数里完成的。而
   release_stripe函数只有将sh->count减到0才会把它挂到某个链表上。所以,这
   里给sh->count加1后,结果总是1。

3.2 handle_stripe
^^^^^^^^^^^^^^^^^^
   通过__get_priority_stripe函数获取到sh之后,接下来就是调用
   handle_stripe来处理这个sh。handle_stripe判断这个sh的raid等级来调相
   应的处理函数。

3.2.1 handle_stripe5
~~~~~~~~~~~~~~~~~~~~~
    首先循环查询一遍每个dev的flag,根据flag的状态设置相应的变量。对于
    stripe0的sh,它的dev0 dev1 dev2对应的flag都是0x04,dev3对应的flag
    是0x00。即,0,1,2三块盘的R5_OVERWRITE被置位。
    循环结束后还要进行许多的状态判断,用来确定这个sh究竟要执行哪些操作。
    最终,实际会被调用到的是handle_stripe_dirtying5函数。

3.2.1.1 handle_stripe_dirtying5
================================
     对于一次写操作,该函数用来判断是要使用rcw还是rmw。比如在一个
     stripe中我只写1块盘。那么我可以通过把要写的这块盘的数据与校验盘的
     数据读回,与新的数据做异或,得出校验数据,这就是rmw。我也可以把所
     有其他盘的数据读回,与新写入的数据做异或,算出校验数据,这就是rcw。
     该函数统计出使用rcw与rmw两种操作时所需的读盘次数,哪种操作需要读
     盘的次数少,就采用哪种操作。
     对于整条带写,肯定是使用rcw,因为一次读盘操作都不需要。

3.2.1.1.1 schedule_reconstruction
----------------------------------
      对于一次回读都不需要的情况,还需调用schedule_reconstruction函数
      来设置一些状态标志。
      对于我们的整条带写操作,会进行如下的设置:
      sh->reconstruct_state = reconstruct_state_drain_run;
      set_bit(STRIPE_OP_BIODRAIN, &s->ops_request);
      set_bit(STRIPE_OP_RECONSTRUCT, &s->ops_request);

3.2.1.2 raid_run_ops
=====================
     接下来,handle_stripe5函数会调用raid_run_ops函数来进行xor运算。在
     raid5.c中有如下宏定义:
     #define raid_run_ops __raid_run_ops
     所以,实际调用的函数是__raid_run_ops。
     该函数首先用memcpy把bio中的数据拷贝到sh的buffer中,然后对sh的
     buffer中的数据进行xor计算。如果硬件支持memcpy和xor操作的话,这些
     操作将会异步进行。在完成后调用callback函数,callback函数中会调用
     release_stripe,以便在适当的时候唤醒raid5d线程进行后续的处理。

3.2.1.3 handle_stripe5结束
===========================
     我们假设使用异步的硬件dma进行memcpy和xor运算,那么,对于整条带的写
     操作,handle_stripe5接下来不会再进行什么实质性的操作了。直到硬件
     操作完成,再次唤醒raid5d后才会处理。

3.3 release_stripe
^^^^^^^^^^^^^^^^^^^
   在raid5d中调用完handle_stripe后,回再次调用release_stripe。由于在
   ops_run_reconstruct5函数中执行了:
   atomic_inc(&sh->count);
   此时sh->count的值是2,减1后得1,不是0,所以不会进行任何操作,直接退
   出。

4 ops_complete_reconstruct
###########################
  memcpy与xor操作都完成后,会调用ops_complete_reconstruct函数。该函数
  会再次调用release_stripe函数,进入release_stripe函数时,sh->count的
  值是1,减1后成为0。因此会再次唤醒raid5d线程。

5 再次进入raid5d
#################
  与第一次执行raid5d一样,依然是先获取sh,调用handle_stripe处理sh,最
  后调用release_stripe。只是这次进入handle_stripe5函数后,会调用
  ops_run_io函数将计算完的校验数据与上层通过make_request传递下来的数据
  一起写入磁盘。

5.1 ops_run_io
^^^^^^^^^^^^^^^
   第一次调用handle_stripe函数的时候也会进入ops_run_io,只是当时
   R5_Wantwrite标志和R5_Wantread都没有被设置,所以不会进行任何操作。然
   而在ops_complete_reconstruct函数中,sh->reconstruct_state的值会被设
   置成reconstruct_state_drain_result。这样,在handle_stripe5函数中,
   就会将所有需要执行写入操作的dev的flag置上R5_Wantwrite标志。
   ops_run_io通过这个标志判断那块盘需要执行写操作。没执行一次写操作前,都
   会把sh->count的值加1。每执行完一块盘的写操作,就会调用一次回调函数
   raid5_end_write_request。该函数会调用release_stripe函数,而
   release_stripe函数会将sh->count的值减1,并检测sh->count是不是已经减
   到0了。。这样,当最后一次写操作完成后,release_stripe函数中会发现
   sh->count的值减到0了,于是第三次唤醒raid5d线程。

6 raid5_end_write_request
##########################
  每一次写磁盘完成后调用,设置一些标志位,并调用release_stripe函数,当
  最后一个写操作完成后,release_stripe函数会唤醒raid5d守护线程。

7 第三次进入raid5d
###################
  与前两次一样,依然是先获取sh,然后处理sh,最后release sh,只是这次在
  handle_stripe5函数中会调用handle_stripe_clean_event函数。

7.1 handle_stripe_clean_event
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
   当一个sh处理完成后,调用该函数设置一些相应的状态。

7.2 release_stripe
^^^^^^^^^^^^^^^^^^^
   sh处理完成,将其挂到inactive_list链表上,并将active_stripes减一。

2010年11月14日星期日

linux内核中的红黑树

和内核中的hash table一样,内核中的红黑树比较“裸”。出于效率方面的考虑,并没有把所有的操作都封装起来。

在执行插入操作的时候,需要使用者自己按照搜索二叉树的方法找到想要插入的节点的父节点,然后调用rb_link_node函数将节点插入,再调用rb_insert_color函数对红黑树进行适当的“旋转”。

而在搜索红黑树的时候,则需要使用者自己按照普通二叉树的搜索方法进行搜索。当然,如何比较键值的大小也是使用者自己决定的。

内核的Documentation目录下有一篇rbtree.txt的文档详细的介绍了该如何使用红黑树。

下面是我写的一个使用红黑树的小例子,可在2.6.32下运行。在这个小程序中,coef被当作键值来使用。

#include <linux/init.h>
#include <linux/module.h>
#include <linux/rbtree.h>

struct q_coef
{
    u8 coef;
    u8 index;
    struct rb_node node;
};

#define COEF_NUM 15
u8 coef[15] = {
    0x01, 0x02, 0x04, 0x08, 0x10, 0x20, 0x40, 0x80,
    0x1d, 0x3a, 0x74, 0xe8, 0xcd, 0x87, 0x13,
};
struct q_coef q_coef[COEF_NUM];

static void q_coef_init(void)
{
    int i;
    memset(&q_coef, 0, sizeof(q_coef));
    for (i = 0 ; i < COEF_NUM ; i++) {
        q_coef[i].coef = coef[i];
        q_coef[i].index = i + 1;
    }
}

struct rb_root q_coef_tree = RB_ROOT;

static int q_coef_insert(struct rb_root *root, struct q_coef *data)
{
    struct rb_node **new = &(root->rb_node), *parent = NULL;

    /* Figure out where to put new code */
    while (*new) {
        struct q_coef *this = rb_entry(*new, struct q_coef, node);
        parent = *new;
        if (data->coef < this->coef)
            new = &((*new)->rb_left);
        else if (data->coef > this->coef)
            new = &((*new)->rb_right);
        else
            return -1;
    }

    /* Add new node and rebalance tree. */
    rb_link_node(&data->node, parent, new);
    rb_insert_color(&data->node, root);

    return 0;
}

static struct q_coef *q_coef_search(struct rb_root *root, u8 coef)
{
    struct rb_node *node = root->rb_node;
    while (node) {
        struct q_coef *data = rb_entry(node, struct q_coef, node);
        if (coef < data->coef)
            node = node->rb_left;
        else if (coef > data->coef)
            node = node->rb_right;
        else
            return data;
    }
    return NULL;
}

static int rbtest_init (void)
{
    int i;
    struct q_coef *ptr;
    struct rb_node *node;
    int ret;

    q_coef_init();

    for (i = 0 ; i < COEF_NUM ; i++) {
        ret = q_coef_insert(&q_coef_tree, &q_coef[i]);
        if (ret < 0) {
            printk(KERN_WARNING "q_coef_insert failed, i=%d\n", i);
            return -1;
        }
    }

    printk(KERN_INFO "search by input order:\n");
    for (i = 0 ; i < COEF_NUM ; i++) {
        ptr = q_coef_search(&q_coef_tree, coef[i]);
        if (ptr == NULL) {
            printk(KERN_WARNING "q_coef_search failed, i=%d\n", i);
            return -1;
        }
        printk(KERN_INFO "coef[%02d]=0x%02x  ptr->coef=0x%02x ptr->index=%02d\n",
            i, coef[i], ptr->coef, ptr->index);
    }

    printk(KERN_INFO "search from first:\n");
    for (node = rb_first(&q_coef_tree) ; node ; node = rb_next(node)) {
        ptr = rb_entry(node, struct q_coef, node);
        printk(KERN_INFO "ptr->coef=0x%02x  ptr->index=%02d\n", ptr->coef, ptr->index);
    }

    printk(KERN_INFO "search from last:\n");
    for (node = rb_last(&q_coef_tree) ; node ; node = rb_prev(node)) {
        ptr = rb_entry(node, struct q_coef, node);
        printk(KERN_INFO "ptr->coef=0x%02x  ptr->index=%02d\n", ptr->coef, ptr->index);
    }

    printk(KERN_INFO "rbtest done\n");
    return -1;
}

static void rbtest_exit (void)
{
}

module_init(rbtest_init);
module_exit(rbtest_exit);

MODULE_LICENSE("Dual BSD/GPL");

2010年11月12日星期五

GF(2**8)的计算器

实现生成多项式为F(x) = x**8 + x**4 + x**3 + x**2 + 1的伽罗华域的加,减,乘,除,乘方运算。可以带括号。

将下面代码保存为pq.py,chmod +x pq.py,然后./pq.py即可进入计算器。按q退出。

解析算数表达式的程序有些问题,不支持乘方和其他运算混合。



#! /usr/bin/env python

# reference to http://www.cnblogs.com/flyingbread/archive/2007/02/03/638932.html

import re

def is_operator(ch):
    if ch == '+' or ch == '-' or ch == '*' or ch == '/' or ch == '**':
        return True
    else:
        return False

def is_parentheses_left(ch):
    if ch == '(':
        return True
    else:
        return False

def is_parentheses_right(ch):
    if ch == ')':
        return True
    else:
        return False

def opt_priority(ch):
    if ch == '+' or ch == '-':
        priority = 1
    elif ch == '*' or ch == '/':
        priority = 2
    elif ch == '**':
        priority = 3
    else:
        # maybe '('
        priority = 0
    return priority

# now we only support 2 operation number
def get_operation_number(ch):
    if ch == '+' or ch == '-' or ch == '*' or ch == '/' or ch == '**':
        return 2
    else:
        return 1

# primitive polynomial for GF(2**8)
# F(x) = x**8 + x**4 + x**3 + x**2 + 1
def do_add(a, b):
    return a ^ b

def do_sub(a, b):
    return a ^ b

gflog = [
    0x00, 0x00, 0x01, 0x19, 0x02, 0x32, 0x1a, 0xc6, 0x03, 0xdf, 0x33, 0xee, 0x1b, 0x68, 0xc7, 0x4b,
    0x04, 0x64, 0xe0, 0x0e, 0x34, 0x8d, 0xef, 0x81, 0x1c, 0xc1, 0x69, 0xf8, 0xc8, 0x08, 0x4c, 0x71,
    0x05, 0x8a, 0x65, 0x2f, 0xe1, 0x24, 0x0f, 0x21, 0x35, 0x93, 0x8e, 0xda, 0xf0, 0x12, 0x82, 0x45,
    0x1d, 0xb5, 0xc2, 0x7d, 0x6a, 0x27, 0xf9, 0xb9, 0xc9, 0x9a, 0x09, 0x78, 0x4d, 0xe4, 0x72, 0xa6,
    0x06, 0xbf, 0x8b, 0x62, 0x66, 0xdd, 0x30, 0xfd, 0xe2, 0x98, 0x25, 0xb3, 0x10, 0x91, 0x22, 0x88,
    0x36, 0xd0, 0x94, 0xce, 0x8f, 0x96, 0xdb, 0xbd, 0xf1, 0xd2, 0x13, 0x5c, 0x83, 0x38, 0x46, 0x40,
    0x1e, 0x42, 0xb6, 0xa3, 0xc3, 0x48, 0x7e, 0x6e, 0x6b, 0x3a, 0x28, 0x54, 0xfa, 0x85, 0xba, 0x3d,
    0xca, 0x5e, 0x9b, 0x9f, 0x0a, 0x15, 0x79, 0x2b, 0x4e, 0xd4, 0xe5, 0xac, 0x73, 0xf3, 0xa7, 0x57,
    0x07, 0x70, 0xc0, 0xf7, 0x8c, 0x80, 0x63, 0x0d, 0x67, 0x4a, 0xde, 0xed, 0x31, 0xc5, 0xfe, 0x18,
    0xe3, 0xa5, 0x99, 0x77, 0x26, 0xb8, 0xb4, 0x7c, 0x11, 0x44, 0x92, 0xd9, 0x23, 0x20, 0x89, 0x2e,
    0x37, 0x3f, 0xd1, 0x5b, 0x95, 0xbc, 0xcf, 0xcd, 0x90, 0x87, 0x97, 0xb2, 0xdc, 0xfc, 0xbe, 0x61,
    0xf2, 0x56, 0xd3, 0xab, 0x14, 0x2a, 0x5d, 0x9e, 0x84, 0x3c, 0x39, 0x53, 0x47, 0x6d, 0x41, 0xa2,
    0x1f, 0x2d, 0x43, 0xd8, 0xb7, 0x7b, 0xa4, 0x76, 0xc4, 0x17, 0x49, 0xec, 0x7f, 0x0c, 0x6f, 0xf6,
    0x6c, 0xa1, 0x3b, 0x52, 0x29, 0x9d, 0x55, 0xaa, 0xfb, 0x60, 0x86, 0xb1, 0xbb, 0xcc, 0x3e, 0x5a,
    0xcb, 0x59, 0x5f, 0xb0, 0x9c, 0xa9, 0xa0, 0x51, 0x0b, 0xf5, 0x16, 0xeb, 0x7a, 0x75, 0x2c, 0xd7,
    0x4f, 0xae, 0xd5, 0xe9, 0xe6, 0xe7, 0xad, 0xe8, 0x74, 0xd6, 0xf4, 0xea, 0xa8, 0x50, 0x58, 0xaf,
]

gfilog = [
    0x01, 0x02, 0x04, 0x08, 0x10, 0x20, 0x40, 0x80, 0x1d, 0x3a, 0x74, 0xe8, 0xcd, 0x87, 0x13, 0x26,
    0x4c, 0x98, 0x2d, 0x5a, 0xb4, 0x75, 0xea, 0xc9, 0x8f, 0x03, 0x06, 0x0c, 0x18, 0x30, 0x60, 0xc0,
    0x9d, 0x27, 0x4e, 0x9c, 0x25, 0x4a, 0x94, 0x35, 0x6a, 0xd4, 0xb5, 0x77, 0xee, 0xc1, 0x9f, 0x23,
    0x46, 0x8c, 0x05, 0x0a, 0x14, 0x28, 0x50, 0xa0, 0x5d, 0xba, 0x69, 0xd2, 0xb9, 0x6f, 0xde, 0xa1,
    0x5f, 0xbe, 0x61, 0xc2, 0x99, 0x2f, 0x5e, 0xbc, 0x65, 0xca, 0x89, 0x0f, 0x1e, 0x3c, 0x78, 0xf0,
    0xfd, 0xe7, 0xd3, 0xbb, 0x6b, 0xd6, 0xb1, 0x7f, 0xfe, 0xe1, 0xdf, 0xa3, 0x5b, 0xb6, 0x71, 0xe2,
    0xd9, 0xaf, 0x43, 0x86, 0x11, 0x22, 0x44, 0x88, 0x0d, 0x1a, 0x34, 0x68, 0xd0, 0xbd, 0x67, 0xce,
    0x81, 0x1f, 0x3e, 0x7c, 0xf8, 0xed, 0xc7, 0x93, 0x3b, 0x76, 0xec, 0xc5, 0x97, 0x33, 0x66, 0xcc,
    0x85, 0x17, 0x2e, 0x5c, 0xb8, 0x6d, 0xda, 0xa9, 0x4f, 0x9e, 0x21, 0x42, 0x84, 0x15, 0x2a, 0x54,
    0xa8, 0x4d, 0x9a, 0x29, 0x52, 0xa4, 0x55, 0xaa, 0x49, 0x92, 0x39, 0x72, 0xe4, 0xd5, 0xb7, 0x73,
    0xe6, 0xd1, 0xbf, 0x63, 0xc6, 0x91, 0x3f, 0x7e, 0xfc, 0xe5, 0xd7, 0xb3, 0x7b, 0xf6, 0xf1, 0xff,
    0xe3, 0xdb, 0xab, 0x4b, 0x96, 0x31, 0x62, 0xc4, 0x95, 0x37, 0x6e, 0xdc, 0xa5, 0x57, 0xae, 0x41,
    0x82, 0x19, 0x32, 0x64, 0xc8, 0x8d, 0x07, 0x0e, 0x1c, 0x38, 0x70, 0xe0, 0xdd, 0xa7, 0x53, 0xa6,
    0x51, 0xa2, 0x59, 0xb2, 0x79, 0xf2, 0xf9, 0xef, 0xc3, 0x9b, 0x2b, 0x56, 0xac, 0x45, 0x8a, 0x09,
    0x12, 0x24, 0x48, 0x90, 0x3d, 0x7a, 0xf4, 0xf5, 0xf7, 0xf3, 0xfb, 0xeb, 0xcb, 0x8b, 0x0b, 0x16,
    0x2c, 0x58, 0xb0, 0x7d, 0xfa, 0xe9, 0xcf, 0x83, 0x1b, 0x36, 0x6c, 0xd8, 0xad, 0x47, 0x8e, 0x00,
]

def do_mul(a, b):
    if a == 0 or b == 0:
        return 0;
    else:
        tmp = gflog[a] + gflog[b]
        tmp = tmp % 255
        return gfilog[tmp]

def do_div(a, b):
    if a == 0:
        return 0
    elif b == 0:
        print 'can not div 0'
        return 0;
    else:
        tmp = gflog[a] - gflog[b]
        if tmp < 0:
            tmp = tmp + 255
        return gfilog[tmp]

def do_power(a, b):
    count = 0
    result = 1
    while (count < b):
        result = do_mul(result, a)
        count += 1
    return result

def calc_once(num, ch):
    if ch == '+':
        ret = do_add(num[1], num[0])
    elif ch == '-':
        ret = do_sub(num[1], num[0])
    elif ch == '*':
        ret = do_mul(num[1], num[0])
    elif ch == '/':
        ret = do_div(num[1], num[0])
    elif ch == '**':
        ret = do_power(num[1], num[0])
    else:
        print 'unknow operation'
        ret = 0
    return ret

# 1. get ch from left to right
# 2. if ch is a number, output it
# 3. if ch is a operator or parenthese:
#    a: if ch is '(', push to stack
#    b: if ch is ')', pop stack until meet '('
#    c: if ch is not parenthese, compare its priority with stack pop
#          if ch priority is higher than the stack pop, push ch to stack
#          else pop stack, push ch to stack
def midfix_to_posfix(midfix):
    stack = []
    posfix = []
    for ch in midfix:
        if is_parentheses_left(ch):
            stack.append(ch)
        elif is_parentheses_right(ch):
            while True:
                ch1 = stack.pop()
                if is_parentheses_left(ch1):
                    break
                else:
                    posfix.append(ch1)
        elif is_operator(ch):
            if len(stack) == 0:
                stack.append(ch)
            else:
                ch1 = stack[-1]
                if opt_priority(ch) > opt_priority(ch1):
                    stack.append(ch)
                else:
                    ch1 = stack.pop()
                    posfix.append(ch1)
                    stack.append(ch)
        else:
            if len(ch) > 2 and ch[0:2] == '0x':
                ch = int(ch, 16)
            else:
                ch = int(ch, 10)
            posfix.append(ch)
    while len(stack) != 0:
        posfix.append(stack.pop())
    return posfix

# get data from left to right
# if ch is a number, push to stack
# if ch is a operator, pop the number it needed, do calc, and push result to stack
# if data is paser comlete, pop the stack as result
def calc_posfix(posfix):
    stack = []
    for ch in posfix:
        if is_operator(ch):
            num = []
            num.append(stack.pop())
            if get_operation_number(ch) > 1:
                num.append(stack.pop())
            stack.append(calc_once(num, ch))
        else:
            stack.append(ch)
    return stack.pop()

def main_loop():
    # match all hex and dec number, and  +,-,*,/,**
    # note: hex must before dec number, and * must before **
    print "please do not use ** mix with other operation, it's not support!"
    regu_for_exp = re.compile('0x[0-9,a-f]+|[0-9]+|\*\*|\*|\+|\-|\/|\(|\)')
    while True:
        expression = raw_input('pq>:')
        if expression != 'quit' and expression != 'q' and expression != 'exit':
            midfix = regu_for_exp.findall(expression)
            posfix = midfix_to_posfix(midfix)
            result = calc_posfix(posfix)
            if result is not None:
                print "0x%02x" % result
            else:
                print result
        else:
            return
if __name__ == '__main__':
    main_loop()

2010年11月10日星期三

raid6中gflog与gfilog

写了一篇介绍raid6中gflog与gfilog的文章,用了太多的数学公式,所以用lyx写了,放到网页上似乎不太方便,于是放到了google code上,下面是下载地址:

http://raid6theory.googlecode.com/files/raid6_theory.pdf

2010年11月3日星期三

创建/sys入口和使用waitqueue的小例子

一个简单的示例程序。创建一个/sys的接口,可以读写,每次回读都是上次写入的内容。每次读写都会触发一次event。

#include <linux/module.h>
#include <linux/kernel.h>
#include <linux/fs.h>
#include <linux/cdev.h>
#include <linux/device.h>
#include <linux/kthread.h>

static struct cdev test_cdev;
static dev_t test_devno;
static struct class *test_class;
struct device *test_device;

struct test_thread {
    wait_queue_head_t    wqueue;
    unsigned long           flags;
    struct task_struct    *tsk;
    unsigned long        timeout;
}test_thread;

#define FROM_SHOW  0
#define FROM_STORE 1
static const struct file_operations test_fops =
{
    .owner          = THIS_MODULE,
};


static int test_fun(void *arg)
{
    struct test_thread *thread = arg;
    allow_signal(SIGKILL);

    while (!kthread_should_stop()) {

        /* We need to wait INTERRUPTIBLE so that
         * we don't add to the load-average.
         * That means we need to be sure no signals are
         * pending
         */
        if (signal_pending(current))
            flush_signals(current);

        wait_event_interruptible_timeout
            (thread->wqueue,
                test_bit(FROM_SHOW, &thread->flags)
                || test_bit(FROM_STORE, &thread->flags)
                || kthread_should_stop(),
                thread->timeout);
        printk("thread %s is waken up by %s\n",
            thread->tsk->comm,
            test_bit(FROM_SHOW, &thread->flags) ? "show" :
            test_bit(FROM_STORE, &thread->flags) ? "store" : "kill");
        thread->flags = 0;
    }

    return 0;
}

#define TEST_LEN  4096
static char test_buf[TEST_LEN];
static ssize_t test_show(struct device *ddev,
            struct device_attribute *attr, char *buf)
{
    int len;
    struct test_thread *thread;

    thread = dev_get_drvdata(ddev);
    len = strlen(test_buf) + 1;
    memcpy(buf, test_buf, len);

    set_bit(FROM_SHOW, &thread->flags);
    wake_up(&thread->wqueue);

    return len;
}

static ssize_t test_store(struct device *ddev,
            struct device_attribute *attr, const char *buf, size_t count)
{
    size_t len;
    struct test_thread *thread;

    thread = dev_get_drvdata(ddev);

    if (count < TEST_LEN - 1)
        len = count;
    else
        len = TEST_LEN - 1;

    memcpy(test_buf, buf, len);
    test_buf[len] = 0;

    set_bit(FROM_STORE, &thread->flags);
    wake_up(&thread->wqueue);

    return count;
}

static DEVICE_ATTR(test1, 0644, test_show, test_store);

static int test_init(void)
{
    int ret;

    ret = alloc_chrdev_region(&test_devno, 0, 255, "test");
    if (ret) {
        printk(KERN_INFO "alloc_chrdev_region failed, ret=%d\n", ret);
        return ret;
    }

    cdev_init(&test_cdev, &test_fops);
    test_cdev.owner = THIS_MODULE;
    ret = cdev_add(&test_cdev, test_devno, 1);
    if (ret) {
        printk(KERN_INFO "cdev_add failed, ret=%d\n", ret);
        goto  free_devno;
    }

    test_class =  class_create(THIS_MODULE, "test_class");
    if (IS_ERR(test_class)) {
        ret = PTR_ERR(test_class);
        printk(KERN_INFO "class_create failed, ret=%d\n", ret);
        goto free_cdev;
    }

    test_device = device_create(test_class, NULL, test_devno, NULL, "test");
    if (IS_ERR(test_device)) {
        ret = PTR_ERR(test_device);
        printk(KERN_INFO "device_create failed, ret=%d\n", ret);
        goto free_class;
    }

    ret = device_create_file(test_device, &dev_attr_test1);
    if (ret) {
        printk(KERN_INFO "device_create_file failed, ret=%d\n", ret);
        goto free_device;
    }

    init_waitqueue_head(&test_thread.wqueue);
    test_thread.timeout = MAX_SCHEDULE_TIMEOUT;
    test_thread.flags = 0;
    dev_set_drvdata(test_device, &test_thread);
    test_thread.tsk = kthread_run(test_fun, &test_thread, "test_thread");
    if (IS_ERR(test_thread.tsk)) {
        ret = PTR_ERR(test_thread.tsk);
        printk(KERN_INFO "kthread_run failed, ret=%d\n", ret);
        goto free_file;
    }

    return 0;

free_file:
    device_remove_file(test_device, &dev_attr_test1);
free_device:
    device_destroy(test_class, test_devno);
free_class:
    class_destroy(test_class);
free_cdev:
    cdev_del(&test_cdev);
free_devno:
    unregister_chrdev_region(test_devno, 255);
    return ret;
}

void test_exit(void)
{
    kthread_stop(test_thread.tsk);
    device_remove_file(test_device, &dev_attr_test1);
    device_destroy(test_class, test_devno);
    class_destroy(test_class);
    cdev_del(&test_cdev);
    unregister_chrdev_region(test_devno, 255);
}
MODULE_LICENSE("GPL");
module_init (test_init);
module_exit (test_exit);

2010年11月1日星期一

在加载驱动时自动创建设备节点

在网上找了些自动创建设备节点的办法,但由于内核接口的变化,已经无法使用了。下面这个程序是可以在2.6.32内核上使用的:


#include <linux/module.h>
#include <linux/kernel.h>
#include <linux/fs.h>
#include <linux/cdev.h>
#include <linux/device.h>

static struct cdev test_cdev;
static dev_t test_devno;
static struct class *test_class;
struct device *test_device;

static const struct file_operations test_fops =
{
    .owner          = THIS_MODULE,
};

static int test_init(void)
{
    int ret;

    ret = alloc_chrdev_region(&test_devno, 0, 255, "test");
    if (ret) {
        printk(KERN_INFO "alloc_chrdev_region failed, ret=%d\n", ret);
        return ret;
    }
    cdev_init(&test_cdev, &test_fops);
    test_cdev.owner = THIS_MODULE;
    ret = cdev_add(&test_cdev, test_devno, 1);
    if (ret) {
        printk(KERN_INFO "cdev_add failed, ret=%d\n", ret);
        goto  free_devno;
    }
    test_class =  class_create(THIS_MODULE, "test_class");
    if (IS_ERR(test_class)) {
        ret = PTR_ERR(test_class);
        printk(KERN_INFO "class_create failed, ret=%d\n", ret);
        goto free_cdev;
    }
    test_device = device_create(test_class, NULL, test_devno, NULL, "test");
    if (IS_ERR(test_device)) {
        ret = PTR_ERR(test_device);
        printk(KERN_INFO "device_create failed, ret=%d\n", ret);
        goto free_class;
    }
    return 0;
free_class:
    class_destroy(test_class);
free_cdev:
    cdev_del(&test_cdev);
free_devno:
    unregister_chrdev_region(test_devno, 255);
    return ret;
}

void test_exit(void)
{
    device_destroy(test_class, test_devno);
    class_destroy(test_class);
    cdev_del(&test_cdev);
    unregister_chrdev_region(test_devno, 255);
}
MODULE_LICENSE("GPL");
module_init (test_init);
module_exit (test_exit);

2010年10月24日星期日

linux 内核 hash table 的使用

很早以前就想学习一下如何使用linux内核中的散列函数。google了几次,发现
网上有大量介绍有关hlist的东西。可找来找去也找不到究竟该如何使用散列。
原来,我先入为主的认为内核中的散列函数会像那些高级语言中实现的散列功能
类似:我提供一对对的(key value)给内核,然后再调用某一个api,传给它一个
key,就可以得到对应的value。

后来参考了一些内核中应用散列的实例才发现,原来根本不是这么回事。实际
上,对于如何将输入数据散列到一个指定范围的算法,需要使用散列的人自己决
定。内核只提供了一个发射碰撞时把碰撞的项链接到一起的hlist结构。

例如,你创建了一个长度为m的散列表,并且已经选择了一个将输入数据映射到
范围0 ~ m-1的散列函数。接下来,你就要在这个长度为m的散列表的每个表项内
放上一个hlist_head结构体。然后在每个输入数据的结构体中定义一个
hlist_node的结构体。每当把一个输入通过散列函数映射到0 ~ m-1的范围内时,就
把这个输入的hlist_node挂到散列表对应的槽的hlist_head上面。当给定一个
key,想获取它的value的时候,就先用散列函数算出这个key对应的槽的位置,
然后遍历这个槽的hlist_node链表,找到与key相等的项。把它的value返回。

例如,有这样一个数组:
0x01, 0x02, 0x04, 0x08,0x10, 0x20, 0x40, 0x80, 0x1d, 0x3a, 0x74, 0xe8,
0xcd, 0x87, 0x13,
其中每个元素对应的索引号为:
1, 2, 3, 4, 5, ... 15
也就是说,当输入0x01时,我希望得到索引号1,当输入0x08时,得到4,当输入
0x3a时,得到10...
这种从数值到索引号的转换,可通过散列来实现。

下面是实现该功能的一个内核代码,散列函数我选择的是:
value = ((104 * key + 52) % 233) % 15
(实际上,对于输入固定的情况,使用完全散列可以获得完全固定的访问时间,
上面这个散列函数就是我想使用完全散列时搜索一个全域散列族得到的第一级散
列函数,但我发先这个散列函数已经足够好,总共才只有一次碰撞。所以就没有
必要像完全散列那样使用二级散列了。)

#include <linux/init.h>
#include <linux/module.h>
#include <linux/list.h>

struct q_coef
{
    u8 coef;
    u8 index;
    struct hlist_node hash;
};

#define HASH_NUMBER 15
u8 coef[HASH_NUMBER] = {
    0x01, 0x02, 0x04, 0x08,0x10, 0x20, 0x40, 0x80, 0x1d, 0x3a, 0x74, 0xe8, 0xcd, 0x87, 0x13,
};
struct q_coef q_coef_list[HASH_NUMBER];

struct hlist_head hashtbl[HASH_NUMBER];

static inline int hash_func(u8 k)
{
    int a, b, p, m;
    a = 104;
    b = 52;
    p = 233;
    m = HASH_NUMBER;
    return ((a * k + b) % p) % m;
}

static void hash_init(void)
{
    int i, j;
    for (i = 0 ; i < HASH_NUMBER ; i++) {
        INIT_HLIST_HEAD(&hashtbl[i]);
        INIT_HLIST_NODE(&q_coef_list[i].hash);
        q_coef_list[i].coef = coef[i];
        q_coef_list[i].index = i + 1;
    }
    for (i = 0 ; i < HASH_NUMBER ; i++) {
        j = hash_func(q_coef_list[i].coef);
        hlist_add_head(&q_coef_list[i].hash, &hashtbl[j]);
    }
}

static void hash_test(void)
{
    int i, j;
    struct q_coef *q;
    struct hlist_node *hn;
    for (i = 0 ; i < HASH_NUMBER ; i++) {
        j = hash_func(coef[i]);
        hlist_for_each_entry(q, hn, &hashtbl[j], hash)
            if (q->coef == coef[i])
                printk("found: coef=0x%02x index=%d\n", q->coef, q->index);
    }
}
static int htest_init (void)
{
    hash_init();
    hash_test();
    return -1;
}

static void htest_exit (void)
{
}

module_init(htest_init);
module_exit(htest_exit);

MODULE_LICENSE("Dual BSD/GPL");