Linux内核网络之网络层接收消息分片组装

　　在接收方，一个由发送方发出的原始IP数据报，其所有分片将被重新组合，然后才能提交给上层协议。
　　而每个将被重新组合的IP数据报都用一个ipq结构实例来表示。structipq｛用来将ipqhash散列表链接成双向链表structhlistnodelist；用来将ipq连接到全局链表ipqlrulist链表中。ipqlrulist用于垃圾收集，当ip组装模块消耗内存大于规定的上限时，会遍历该链表清除符合条件的分片structlistheadlrulist；lrulistmemberu32user；标识分片来源：来自网络其他主机或是本地环回接口的分片、含有路由警告选项的IP分片以下四个字段的值都来源于ip首部，用来唯一确定分片来自哪个ip数据报be32saddr；be32daddr；be16id；u8protocol；u8lastin；defineCOMPLETE4所有分片已到达，可以进行组装；defineFIRSTIN2第一个分片到达，其特殊之处在于只有第一个分片包含了所有ip选项；defineLASTIN1最后一个分片已到达，最后一个分片带有原始数据包的长度信息用来链接已经接收到的分片structskbufffragments；linkedlistofreceivedfragments当前已接收到分片中offset最大的那个分片的offset值加上其长度值，即分片末尾处在整个原始数据报中的位置，因此当收到最后一个分片后该字段值将更新为原始数据报的长度intlen；totallengthoforiginaldatagram已接收到的所有分片总长度，因此可以用len和meat来判断一个ip数据报的所有分片时否已到齐intmeat；自旋锁，在smp环境下，处理ipq及分片链时需上锁spinlocktlock；atomictrefcnt；引用计数组装超时定时器，组装分片非常消耗资源，防止无休止等待分片的到达structtimerlisttimer；whenwillthisqueueexpire？记录最后一个分片的到达时间，在组装数据报时用该值作为时间戳structtimevalstamp；接收最后一个分片的网络设备索引号。当分片组装失败时，用该设备发送组装失败icmp出错报文。intiif；已接收到分片的计数器，可通过对端信息块peer中的分片计数器和该分片计数器来防止DoS攻击unsignedintrid；记录发送方的一些信息structinetpeerpeer；｝；Hashtable。defineIPQHASHSZ64Perbucketlockiseasytoaddnow。staticstructhlistheadipqhash〔IPQHASHSZ〕；
　　每个原始的IP数据报的所有分片以链表的形式保存在ipq结构fragments中。
　　在网络层中，会根据每个原始的IP数据报首部中的（saddr，daddr，id，protocol，ipfraghashrnd）计算一个hash值，然后将ipq结构放到对应的ipqhash〔hash〕散列表中。
　　因此，当一个新的分片skb到来时，根据（saddr，daddr，id，protocol，ipfraghashrnd）计算出hash值，从ipqhash〔hash〕散列表中找到ipq结构，然后把分片存放到fragments链表中。
　　当IP分片到达本地时，先调用ipdefrag进行重组。ip数据报输入到本地intiplocaldeliver（structskbuffskb）｛ReassembleIPfragments。若接收到的ip报是分片，则调用ipdefrag进行重组，其标志为IPDEFRAGLOCALDELIVERif（skbnh。iphfragoffhtons（IPMFIPOFFSET））｛返回0，表示ip数据包分片还未到齐，重组未完成，直接返回；非0，返回已完成重组的ip数据报skb缓存的指针，传到传输层进行处理skbipdefrag（skb，IPDEFRAGLOCALDELIVER）；if（！skb）return0；｝通过netfilter处理后，调用iplocaldeliverfinish将组装完的ip报文传送到传输层进行处理returnNFHOOK（PFINET，NFIPLOCALIN，skb，skbdev，NULL，iplocaldeliverfinish）；｝
　　分片组装
　　分片组装的流程如下：
　　ipdefrag
　　ipfind在ipq散列表中查找分片所属的ipq，若找不到则新建一个ipq
　　ipfragqueue将分片插入到ipq分片链表的适当位置
　　ipfragreasm原始数据报的所有分片全部到达，组装分片
　　具体实现细节如下
　　ipdefragProcessanincomingIPdatagramfragment。对分片进行组装skb：新接收到的ip数据报user：分片来源structskbuffipdefrag（structskbuffskb，u32user）｛structiphdriphskbnh。iph；structipqqp；structnetdevicedev；IPINCSTATSBH（IPSTATSMIBREASMREQDS）；Startbycleaningupthememory。若iqp散列表消耗的内存大于指定的值，则ipevictor（）清理分片if（atomicread（ipfragmem）sysctlipfraghighthresh）ipevictor（）；获取接收数据报的网络设备指针devskbdev；Lookup（orcreate）queueheader在ipq散列表中查找分片所属的ipq。若找不到则新建一个ipq，若返回为null，则说明查找及创建失败if（（qpipfind（iph，user））！NULL）｛structskbuffretNULL；spinlock（qplock）；将分片插入到ipq分片链表的适当位置ipfragqueue（qp，skb）；当ipq的第一个分片和最后一个分片都已收到，且若已接收数据报的总长度与原始数据报的长度相等，则说明该原始数据报的所有分片都已到齐if（qplastin（FIRSTINLASTIN）qpmeatqplen）调用ipfragreasm组装分片retipfragreasm（qp，dev）；spinunlock（qplock）；删除iqp及其所有分片ipqput（qp，NULL）；returnret；｝IPINCSTATSBH（IPSTATSMIBREASMFAILS）；释放分片kfreeskb（skb）；returnNULL；｝
　　当一个分片到达后，按照偏移量插入到ipq分片链表的适当位置，如下图：
　　ipfind
　　在ipq散列表中查找分片所属的ipq结构如下：根据分片的ip首部以及user标志在ipq散列表中找对应的ipq，若没找到，则为其创建新的ipqstaticinlinestructipqipfind（structiphdriph，u32user）｛be16idiphid；be32saddriphsaddr；be32daddriphdaddr；u8protocoliphprotocol；unsignedinthash；structipqqp；structhlistnoden；readlock（ipfraglock）；计算hash值hashipqhashfn（id，saddr，daddr，protocol）；遍历该桶的ipq链表查找对应的ipq，若找到则返回该ipq，否则说明这是一个新的ip数据报分片，需要创建一个新的ipqhlistforeachentry（qp，n，ipqhash〔hash〕，list）｛if（qpididqpsaddrsaddrqpdaddrdaddrqpprotocolprotocolqpuseruser）｛atomicinc（qprefcnt）；readunlock（ipfraglock）；returnqp；｝｝readunlock（ipfraglock）；返回新建的ipqreturnipfragcreate（iph，user）；｝
　　ipfragcreate
　　创建ipq结构，并初始化其组装超时定时器。每当接收到一个属于新的ip数据报的分片时，会为其创建对应的iqp，并初始化其组装超时定时器staticstructipqipfragcreate（structiphdriph，u32user）｛structipqqp；if（（qpfragallocqueue（））NULL）gotooutnomem；qpprotocoliphprotocol；qplastin0；qpidiphid；qpsaddriphsaddr；qpdaddriphdaddr；qpuseruser；qplen0；qpmeat0；qpfragmentsNULL；qpiif0；qppeersysctlipfragmaxdist？inetgetpeer（iphsaddr，1）：NULL；Initializeatimerforthisentry。inittimer（qptimer）；qptimer。data（unsignedlong）qp；pointertoqueueqptimer。functionipexpire；expirefunctionspinlockinit（qplock）；atomicset（qprefcnt，1）；returnipfragintern（qp）；outnomem：LIMITNETDEBUG（KERNERRipfragcreate：nomemoryleft！）；returnNULL；｝
　　ipfragintern将新建的ipq插入到ipq散列表中和ipqlrulist中staticstructipqipfragintern（structipqqpin）｛structipqqp；unsignedinthash；writelock（ipfraglock）；计算hash值hashipqhashfn（qpinid，qpinsaddr，qpindaddr，qpinprotocol）；qpqpin；安装ipq的组装超时定时器，定时为sysctlipfragtimeif（！modtimer（qptimer，jiffiessysctlipfragtime））atomicinc（qprefcnt）；递增ipq的引用计数atomicinc（qprefcnt）；将ipq插入到ipq散列表和lrulist中hlistaddhead（qplist，ipqhash〔hash〕）；INITLISTHEAD（qplrulist）；listaddtail（qplrulist，ipqlrulist）；对ipq的数量进行计数ipfragnqueues；writeunlock（ipfraglock）；returnqp；｝
　　ipfragqueueAddnewsegmenttoexistingqueue。将分片skb添加到iqp指定的ipq分片链表中qp：将添加ip分片到ipqskb：接收到待添加到ipq中的ip分片staticvoidipfragqueue（structipqqp，structskbuffskb）｛structskbuffprev，next；intflags，offset；intihl，end；对分片已全部接收到的ipq，则释放该分片后返回if（qplastinCOMPLETE）gotoerr；若不是有本地生成的分片，则调用ipfragtoofar检测该分片是否存在DoS攻击嫌疑，。若受到攻击，则调用ipfragreinit释放所有的ipq所有分片if（！（IPCB（skb）flagsIPSKBFRAGCOMPLETE）unlikely（ipfragtoofar（qp））unlikely（ipfragreinit（qp）））｛ipqkill（qp）；gotoerr；｝取出ip首部中的标志位、片偏移及首部长度字段，并计算片偏移值和首部长度值offsetntohs（skbnh。iphfragoff）；flagsoffsetIPOFFSET；ip首部中的片偏移字段为13位，表示的是8字节的倍数offsetIPOFFSET；offset3；offsetisin8bytechunksihlskbnh。iphihl4；Determinethepositionofthisfragment。计算分片末尾处在原始数据报中的位置endoffsetskblenihl；Isthisthefinalfragment？若是最后一个分片if（（flagsIPMF）0）｛Ifwealreadyhavesomebitsbeyondendorhavedifferentend，thesegmentiscorrrupted。先对分片进行检查，若其末尾小于原始包长度，或者ipq已有LASTIN标志且分片末尾不等于原始包长度，则出错if（endqplen（（qplastinLASTIN）end！qplen））gotoerr；设置LASTIN标志，将完整数据报长度存储在ipq的len字段中qplastinLASTIN；qplenend；｝else｛不是最后一个分片，其数据长度又不是8字节对齐，则将其截为8字节对齐if（end7）｛end7；若需要计算校验和，则强制设置有软件来计算校验和，因为截断了ip有效负载，改变了长度，需重新计算校验和if（skbipsummed！CHECKSUMUNNECESSARY）skbipsummedCHECKSUMNONE；｝在最后一个分片没有到达的情况下，若当前分片的末尾在整个数据报中的位置大于ipq中的len值，则更新len。因为ipq中的len字段始终保持所有已接收到的分片中分片末尾在数据报中的位置的最大值，而只有在收到最后一个分片后。len值才是整个数据报的长度if（endqplen）｛Somebitsbeyondendcorruption。若此数据报有异常，则直接丢弃if（qplastinLASTIN）gotoerr；qplenend；｝｝若分片的数据区长度为0，则该分片异常，直接丢弃if（endoffset）gotoerr；调用pskbpull去掉ip首部，只保留数据部分if（pskbpull（skb，ihl）NULL）gotoerr；将skb数据区长度调整为endoffset，ip有效负载长度if（pskbtrimrcsum（skb，endoffset））gotoerr；Findoutwhichfragmentsareinfrontandatthebackofusinthechainoffragmentssofar。Wemustknowwheretoputthisfragment，right？确定分片在分片链表中的位置。因为各分片很可能不按照顺序到达，而ipq分片链表上的分片是按照分片偏移值从小到大的顺序连接在一起的prevNULL；for（nextqpfragments；next！NULL；nextnextnext）｛if（FRAGCB（next）offsetoffset）break；bingo！prevnext；｝Wefoundwheretoputthisone。Checkforoverlapwithprecedingfragment，and，ifneeded，alignthingssothatanyoverlapsareeliminated。检测和上一个分片的数量是否有重叠，i是重叠部分的长度，若有重叠，调用pskbpull去掉重叠的部分if（prev）｛inti（FRAGCB（prev）offsetprevlen）offset；if（i0）｛offseti；if（endoffset）gotoerr；if（！pskbpull（skb，i））gotoerr；if（skbipsummed！CHECKSUMUNNECESSARY）skbipsummedCHECKSUMNONE；｝｝若和后一个分片有重叠，则还需判断重叠部分的数据长度是否超过下一个分片的数据长度，若没有超过则调整下一个分片，超过这需要释放下一个分片后再检查与后面第二个分片的数据是否有重叠，如此反复，直到完成后面对所有分片的检测。调整分片的偏移值、已接收分片总长度等while（nextFRAGCB（next）offsetend）｛intiendFRAGCB（next）offset；overlapisibytesif（inextlen）｛Eatheadofthenextoverlappedfragmentandleavetheloop。Thenextonescannotoverlap。if（！pskbpull（next，i））gotoerr；FRAGCB（next）offseti；qpmeati；if（nextipsummed！CHECKSUMUNNECESSARY）nextipsummedCHECKSUMNONE；break；｝else｛structskbufffreeitnext；Oldfragmentiscompletelyoverriddenwithnewonedropit。nextnextnext；if（prev）prevnextnext；elseqpfragmentsnext；qpmeatfreeitlen；fragkfreeskb（freeit，NULL）；｝｝记录当前分片的偏移值FRAGCB（skb）offsetoffset；Insertthisfragmentinthechainoffragments。将当前的分片插入到ipq分片队列中的相应位置skbnextnext；if（prev）prevnextskb；elseqpfragmentsskb；if（skbdev）qpiifskbdevifindex；skbdevNULL；更新ipq的时间戳skbgettimestamp（skb，qpstamp）；累计该分片已收到的分片总长度qpmeatskblen；累计分片组装模块所占的内存atomicadd（skbtruesize，ipfragmem）；若片偏移值为0，说明当前分片为第一个分片，设置FIRSTINif（offset0）qplastinFIRSTIN；writelock（ipfraglock）；调整所属ipq在ipqlrulist中的位置，这是为了在占用内存超过阈值时可以先释放最久未用的那些分片listmovetail（qplrulist，ipqlrulist）；writeunlock（ipfraglock）；return；err：kfreeskb（skb）；｝
　　对于其中计算分片末尾处在原始数据报中的位置的地方
　　计算分片末尾处在原始数据报中的位置
　　endoffsetskblenihl；
　　有关ihl、offset、len和end的关系，如下图：
　　ipfragreasm组装已到齐的所有分片。当原始数据报所有分片都已到齐时，调用此函数组装分片qp：存储待组装分片队列的ipqdev：输入分片的网络设备staticstructskbuffipfragreasm（structipqqp，structnetdevicedev）｛structiphdriph；structskbufffp，headqpfragments；intlen；intihlen；组装前，先将此ipq节点从ipq散列表和ipqlrulist表中断开，并删除定时器ipqkill（qp）；BUGTRAP（head！NULL）；BUGTRAP（FRAGCB（head）offset0）；Allocateanewbufferforthedatagram。计算原始数据报包括ip首部的总长度，ihlenheadnh。iphihl4；lenihlenqplen；若该长度值超过64k则丢弃if（len65535）gotooutoversize；Headoflistmustnotbecloned。在组装分片时，所有的分片都会组装到第一个分片上，因此第一个分片不能时克隆的，若是克隆的，则需为分片组装重新分配一个skbif（skbcloned（head）pskbexpandhead（head，0，0，GFPATOMIC））gotooutnomem；Ifthefirstfragmentisfragmenteditself，wesplitittotwochunks：thefirstwithdataandpagedpartandthesecond，holdingonlyfragments。分片队列的第一个skb不能既带有数据，又带有分片，也即是其fraglist上不能有分片skb，若有则重新分配一个skb。最终效果时，head自身不包括数据，其fraglist上连接着所有分片的skb。这也是skb的一种表现形式，不一定是一个连续的数据块，但最终会调用skblinearize（）将这些数据都复制到一个连续的数据块中if（skbshinfo（head）fraglist）｛structskbuffclone；inti，plen0；if（（cloneallocskb（0，GFPATOMIC））NULL）gotooutnomem；clonenextheadnext；headnextclone；skbshinfo（clone）fraglistskbshinfo（head）fraglist；skbshinfo（head）fraglistNULL；for（i0；iskbshinfo（head）nrfrags；i）plenskbshinfo（head）frags〔i〕。size；clonelenclonedatalenheaddatalenplen；headdatalenclonelen；headlenclonelen；clonecsum0；cloneipsummedheadipsummed；atomicadd（clonetruesize，ipfragmem）；｝把所有分片组装起来，即将分片连接到第一个skb中的fraglist上，同时还需要遍历所有分片，重新计算ip数据报长度以及校验和等。skbshinfo（head）fraglistheadnext；skbpush（head，headdataheadnh。raw）；atomicsub（headtruesize，ipfragmem）；for（fpheadnext；fp；fpfpnext）｛headdatalenfplen；headlenfplen；if（headipsummed！fpipsummed）headipsummedCHECKSUMNONE；elseif（headipsummedCHECKSUMCOMPLETE）headcsumcsumadd（headcsum，fpcsum）；headtruesizefptruesize；atomicsub（fptruesize，ipfragmem）；｝headnextNULL；headdevdev；skbsettimestamp（head，qpstamp）；重置首部长度、片偏移、标志位和总长度iphheadnh。iph；iphfragoff0；iphtotlenhtons（len）；IPINCSTATSBH（IPSTATSMIBREASMOKS）；既然各分片都已处理完，释放ipq的分片队列qpfragmentsNULL；returnhead；outnomem：LIMITNETDEBUG（KERNERRIP：queueglue：nomemoryforgluingqueuep，qp）；gotooutfail；outoversize：if（netratelimit（））printk（KERNINFOOversizedIPpacketfromd。d。d。d。，NIPQUAD（qpsaddr））；outfail：IPINCSTATSBH（IPSTATSMIBREASMFAILS）；returnNULL；｝
　　组装的过程就是把所有分片组装起来，即将分片连接到第一个skb中的fraglist上，并返回组装后的数据报文。
　　ipq散列表的重组
　　所有的分片重组都是通过ipq散列表进行的。随着后续ipq的添加或删除，使得散列表中的ipq的分布变得不均匀，处理性能会大大降低，因此需要定时对散列表进行重新组装。这样做同时也是为了防御DoS攻击。
　　散列表的定时重组是通过ipfragsecrettimer定时器实现的，在ipfraginit（）中对ipfragsecrettimer定时器的初始化。在该函数中还初始化了ipfraghashrnd变量，该变量主要用来与IP首部中的源地址、目的地址等构成ipq散列表的关键字。每次重组时都会将ipfraghashrnd更新为一个新的随机值，并重新设置ipfragsecrettimer定时器，时间跨度为10min。
　　ipfraginitvoidipfraginit（void）｛ipfraghashrnd（u32）（（numphyspages（numphyspages7））（jiffies（jiffies6）））；inittimer（ipfragsecrettimer）；ipfragsecrettimer。functionipfragsecretrebuild；ipfragsecrettimer。expiresjiffiessysctlipfragsecretinterval；addtimer（ipfragsecrettimer）；｝
　　ipfragsecretrebuild对全局的ipq散列表进行重组staticvoidipfragsecretrebuild（unsignedlongdummy）｛unsignedlongnowjiffies；inti；writelock（ipfraglock）；重新获取ipfraghashrnd随机值getrandombytes（ipfraghashrnd，sizeof（u32））；遍历ipq散列表中所有ipq，根据新的ipfraghashrnd值把这些ipq重新连接到散列表对应的桶中for（i0；iIPQHASHSZ；i）｛structipqq；structhlistnodep，n；hlistforeachentrysafe（q，p，n，ipqhash〔i〕，list）｛unsignedinthvalipqhashfn（qid，qsaddr，qdaddr，qprotocol）；if（hval！i）｛hlistdel（qlist）；Relinktonewhashchain。hlistaddhead（qlist，ipqhash〔hval〕）；｝｝｝writeunlock（ipfraglock）；重新设置重构定时器的下次到期时间modtimer（ipfragsecrettimer，nowsysctlipfragsecretinterval）；｝
　　清除超时的IP分片
　　在复杂的网络环境下，一个IP数据报的分片有可能不能全部抵达目的地址，而该数据报已到达的分片会占用大量的资源，此外也为了防止抵御DoS攻击，因此需要设置一个时钟，一旦超时，数据报的分片还未全部到达，则将其已到达的分片全部清除。
　　每当收到一个属于新的IP数据报分片时，在为其创建ipq时，会初始化其超时定时器ipexpire每当接收到一个属于新的ip数据报的分片时，会为其创建对应的iqp，并初始化其组装超时定时器staticstructipqipfragcreate（structiphdriph，u32user）｛。。。qptimer。functionipexpire；expirefunction。。。｝
　　ipexpire组装超时定时器例程，当定时器激活时，清除在规定时间内没有完成组装的ipq及其所有分片staticvoidipexpire（unsignedlongarg）｛structipqqp（structipq）arg；spinlock（qplock）；当前已是COMPLETE状态，不做处理，直接跳到释放ipq及其所有的分片处if（qplastinCOMPLETE）gotoout；将ipq从ipq散列表和ipqlrulist链表中删除ipqkill（qp）；IPINCSTATSBH（IPSTATSMIBREASMTIMEOUT）；IPINCSTATSBH（IPSTATSMIBREASMFAILS）；若第一个分片已经到达，则发送分片组装超时ICMP出错报文if（（qplastinFIRSTIN）qpfragments！NULL）｛structskbuffheadqpfragments；SendanICMPFragmentReassemblyTimeoutmessage。if（（headdevdevgetbyindex（qpiif））！NULL）｛icmpsend（head，ICMPTIMEEXCEEDED，ICMPEXCFRAGTIME，0）；devput（headdev）；｝｝out：spinunlock（qplock）；释放ipq及其所有的ip分片ipqput（qp，NULL）；｝
　　垃圾收集
　　为了控制IP组装所占用的内存，设置了两个阈值ipfraghighthresh和ipfraglowthresh。当前ipq散列表占用的内存量存储在全局变量ipfragmem中，当ipfragmem大于ipfraghighthresh时，需要调用ipevictor（）对散列表进行清理，直到ipfragmem降低到ipfraglowthresh。这两个阈值可以在系统运行时通过proc文件系统修改。
　　ipevictor
　　该方法主要对ipq中的分片进行条件性的清理。在所有的ipq中，若分片没有到齐，则被删除。staticvoidipevictor（void）｛structipqqp；structlistheadtmp；intwork；在清理前再次对当前消耗的内存量做测试，若少于sysctlipfraglowthresh，则不进行清理workatomicread（ipfragmem）sysctlipfraglowthresh；if（work0）return；while（work0）｛readlock（ipfraglock）；若lru链表为空，解锁后返回if（listempty（ipqlrulist））｛readunlock（ipfraglock）；return；｝tmpipqlrulist。next；qplistentry（tmp，structipq，lrulist）；递增ipq引用计数atomicinc（qprefcnt）；readunlock（ipfraglock）；在删除分片前后要做同步保护spinlock（qplock）；若分片还未到齐，则ipq从散列表及lru链表中删除。ipqkill只删除不释放if（！（qplastinCOMPLETE））ipqkill（qp）；spinunlock（qplock）；ipqput真正删除ipq及其所有分片ipqput（qp，work）；IPINCSTATSBH（IPSTATSMIBREASMFAILS）；｝｝