Linux内核中的随机早期检测算法(RED)

独孤九贱

好久都没有来发过贴子了，由于需要，看了一下内核中RED的实现。发上来共享之：
内核版本2.6.12，当时对比过2.6.22，22版的几本上一样，就是多了些封装函数。12要简洁些，故分析这个版本。其中有些地方还没有完全吃透，等分析明白了再补上。发先上来大家讨论：

算法描述：
http://www.zlunwen.com/computer/application/27228.htm

说得很详细了。
其中关键环节在于：

当有分组到达时：

If( 队列空)
{
    =f(time-q_time);
   Avq= (1-w)m Avq;
}
else
   Avq<—(1－w) Avq+wq；

If (MINth<= Avq <=MAXth)
{
   Count=Count+1;
   Pb＝maxp((Avq- MINth)/( MAXth- MINth));

    Pa=Pb/ (1-count*P);
}
else if(Avq>= MAXth)
   丢弃分组;
else
  count=-1;

算法之关键，在于求得一个“平均队列长度(average length queue)”，然后跟两个预设阀值比较(MAXth,MINth)比较，分别处理
< MINth
[MIN,MAXth]
> MAXth
三种情况。
这其中还要注意两点：
1、而在计算平均队列长度的时候，需要对队列是否为空进行区别处理。
2、在[MINth,MAXth]中进行随机丢包时，需要按公式计算出一个随机丢包的概率。

实现分析：

struct red_sched_data
{
/* Parameters */
        u32                limit;                /* 队列长度上限        */
        u32                qth_min;                        /* 算法需要的两个两个门限值*/
        u32                qth_max;       
        u32                Rmask;
        u32                Scell_max;                /* 最大空闲时间 */
        unsigned char        flags;
        char                Wlog;                /* log(W)                */
        char                Plog;                /* random number bits        */
        char                Scell_log;
        u8                Stab[256];

/* Variables */
        unsigned long        qave;                /* 这个东东太重要了，平均队列长度 */
        int                qcount;                /* 上次丢弃分组后收到的分组个数*/
        u32                qR;                /* Cached random number */

        psched_time_t        qidlestart;        /* 队列空间开始时间                */
        struct tc_red_xstats st;
};

struct red_sched_data是RED的私有数据结构，包含了其所有需要的重要参数。

如前所述，算法中需要处理队列是否为空的情况，上述数据结构中，使用qidlestart成员来描述队列是否空：

1. 如果为空，则记录空闲开始时间，通过宏PSCHED_GET_TIME(q->qidlestart);来完成；
   
2. 如果不为空，则置为0，         通过宏PSCHED_SET_PASTPERFECT(q->qidlestart);来完成；

复制代码

例如，在出队函数中：

1. static struct sk_buff *
   
2. red_dequeue(struct Qdisc* sch)
   
3. {
   
4.         struct sk_buff *skb;
   
5.         struct red_sched_data *q = qdisc_priv(sch);
   
6. 
   
7.         skb = __skb_dequeue(&sch->q);
   
8.         if (skb) {
   
9. ……
   
10.         }
    
11.         [color=Red]PSCHED_GET_TIME(q->qidlestart);[/color]
    
12.         return NULL;
    
13. }

复制代码

如果从队列中再也取不到数据包（即队列为空，空闲了），则调用PSCHED_GET_TIME(q->qidlestart)宏将空闲开始时间设为当前时间。

同理，在数据包重入队函数中：

1. static int
   
2. red_requeue(struct sk_buff *skb, struct Qdisc* sch)
   
3. {
   
4.         struct red_sched_data *q = qdisc_priv(sch);
   
5. 
   
6.         PSCHED_SET_PASTPERFECT(q->qidlestart);

复制代码

因为有数据包进入队列，队列不在空闲，则将qidlestart置为0。

因为这个东东在程序中反复用到，所以先将其分析了。

一、注册
首先，模块向QoS子系统注册自身，这一块实现跟QoS模块关系紧密，以后再来单独分析：

1. static int __init red_module_init(void)
   
2. {
   
3.         return register_qdisc(&red_qdisc_ops);
   
4. }

复制代码

简单地说，每一个QoS算法，都被封装成一个对像 struct Qdisc_ops：

1. static struct Qdisc_ops red_qdisc_ops = {
   
2.         .next                =        NULL,
   
3.         .cl_ops                =        NULL,
   
4.         .id                =        "red",
   
5.         .priv_size        =        sizeof(struct red_sched_data),
   
6.         .enqueue        =        red_enqueue,
   
7.         .dequeue        =        red_dequeue,
   
8.         .requeue        =        red_requeue,
   
9.         .drop                =        red_drop,
   
10.         .init                =        red_init,
    
11.         .reset                =        red_reset,
    
12.         .change                =        red_change,
    
13.         .dump                =        red_dump,
    
14.         .dump_stats        =        red_dump_stats,
    
15.         .owner                =        THIS_MODULE,
    
16. };

复制代码

它封装了算法的一些属性和操作。

二、初始化
初始化函数是red_init：

1. static int red_init(struct Qdisc* sch, struct rtattr *opt)
   
2. {
   
3.         return red_change(sch, opt);
   
4. }

复制代码

red_change用来设置RED的属性值：

1. static int red_change(struct Qdisc *sch, struct rtattr *opt)
   
2. {
   
3.         struct red_sched_data *q = qdisc_priv(sch);
   
4.         struct rtattr *tb[TCA_RED_STAB];
   
5.         struct tc_red_qopt *ctl;
   
6. 
   
7.         if (opt == NULL ||
   
8.             rtattr_parse_nested(tb, TCA_RED_STAB, opt) ||
   
9.             tb[TCA_RED_PARMS-1] == 0 || tb[TCA_RED_STAB-1] == 0 ||
   
10.             RTA_PAYLOAD(tb[TCA_RED_PARMS-1]) < sizeof(*ctl) ||
    
11.             RTA_PAYLOAD(tb[TCA_RED_STAB-1]) < 256)
    
12.                 return -EINVAL;
    
13. 
    
14.         ctl = RTA_DATA(tb[TCA_RED_PARMS-1]);
    
15. 
    
16.         sch_tree_lock(sch);
    
17.         q->flags = ctl->flags;
    
18.         q->Wlog = ctl->Wlog;
    
19.         q->Plog = ctl->Plog;
    
20.         q->Rmask = ctl->Plog < 32 ? ((1<<ctl->Plog) - 1) : ~0UL;
    
21.         q->Scell_log = ctl->Scell_log;
    
22.         q->Scell_max = (255<<q->Scell_log);
    
23.         q->qth_min = ctl->qth_min<<ctl->Wlog;
    
24.         q->qth_max = ctl->qth_max<<ctl->Wlog;
    
25.         q->limit = ctl->limit;
    
26.         memcpy(q->Stab, RTA_DATA(tb[TCA_RED_STAB-1]), 256);
    
27. 
    
28.         q->qcount = -1;
    
29.         if (skb_queue_len(&sch->q) == 0)
    
30.                 PSCHED_SET_PASTPERFECT(q->qidlestart);
    
31.         sch_tree_unlock(sch);
    
32.         return 0;
    
33. }

复制代码

没有仔细去看过iproute2设置RED的代码，如果一对照，应该很好理解相关的参数。

二、入队函数
前面前戏整了一大串，现在写到重点了，整个算法的实现，是在入队函数中完成的，函数有点长，一步步地来。

1.         struct red_sched_data *q = qdisc_priv(sch);
   
2. 
   
3.         psched_time_t now;
   
4. 
   
5.         if (!PSCHED_IS_PASTPERFECT(q->qidlestart)) {
   
6.                 long us_idle;
   
7.                 int  shift;
   
8. 
   
9.                 PSCHED_GET_TIME(now);
   
10.                 us_idle = PSCHED_TDIFF_SAFE(now, q->qidlestart, q->Scell_max);
    
11.                 PSCHED_SET_PASTPERFECT(q->qidlestart);
    
12. 
    
13. /*
    
14.    The problem: ideally, average length queue recalcultion should
    
15.    be done over constant clock intervals. This is too expensive, so that
    
16.    the calculation is driven by outgoing packets.
    
17.    When the queue is idle we have to model this clock by hand.
    
18. 
    
19.    SF+VJ proposed to "generate" m = idletime/(average_pkt_size/bandwidth)
    
20.    dummy packets as a burst after idle time, i.e.
    
21. 
    
22.           q->qave *= (1-W)^m
    
23. 
    
24.    This is an apparently overcomplicated solution (f.e. we have to precompute
    
25.    a table to make this calculation in reasonable time)
    
26.    I believe that a simpler model may be used here,
    
27.    but it is field for experiments.
    
28. */
    
29.                 shift = q->Stab[us_idle>>q->Scell_log];
    
30. 
    
31.                 if (shift) {
    
32.                         q->qave >>= shift;
    
33.                 } else {
    
34.                         /* Approximate initial part of exponent
    
35.                            with linear function:
    
36.                            (1-W)^m ~= 1-mW + ...
    
37. 
    
38.                            Seems, it is the best solution to
    
39.                            problem of too coarce exponent tabulation.
    
40.                          */
    
41. 
    
42.                         us_idle = (q->qave * us_idle)>>q->Scell_log;
    
43.                         if (us_idle < q->qave/2)
    
44.                                 q->qave -= us_idle;
    
45.                         else
    
46.                                 q->qave >>= 1;
    
47.                 }
    
48.         }

复制代码

首先根据qidlestart判断队列如果是空间的话，则取得当前时间，并计算空闲时间总长度，然后置队列非空闲标记（因为有数据进入了）
                PSCHED_GET_TIME(now);
                us_idle = PSCHED_TDIFF_SAFE(now, q->qidlestart, q->Scell_max);
                PSCHED_SET_PASTPERFECT(q->qidlestart);
如前所述，队列空间时，平均队列长度的计算公式为：

q->qave *= (1-W)^m

shift的目的是为了计算m：
shift = q->Stab[us_idle>>q->Scell_log];
当shift求出后，就可以根据公式，计算平均队列长度了：

1.                 if (shift) {
   
2.                         q->qave >>= shift;
   
3.                 } else {
   
4.                         /* Approximate initial part of exponent
   
5.                            with linear function:
   
6.                            (1-W)^m ~= 1-mW + ...
   
7. 
   
8.                            Seems, it is the best solution to
   
9.                            problem of too coarce exponent tabulation.
   
10.                          */
    
11. 
    
12.                         us_idle = (q->qave * us_idle)>>q->Scell_log;
    
13.                         if (us_idle < q->qave/2)
    
14.                                 q->qave -= us_idle;
    
15.                         else
    
16.                                 q->qave >>= 1;
    
17.                 }

复制代码

如果队列非空，则计算公式为：
avg = (1-W)*avg + W*current_queue_len
也就是
qave = qave*(1-W) + sch->qstats.backlog*W;
也就是
q->qave += sch->qstats.backlog - (q->qave >> q->Wlog);


接下来比较平均队列跟阀值的关系，如果小于MINth的话：

1.         if (q->qave < q->qth_min) {
   
2.                 q->qcount = -1;
   
3. enqueue:
   
4.                 if (sch->qstats.backlog + skb->len <= q->limit) {
   
5.                         __skb_queue_tail(&sch->q, skb);
   
6.                         sch->qstats.backlog += skb->len;
   
7.                         sch->bstats.bytes += skb->len;
   
8.                         sch->bstats.packets++;
   
9.                         return NET_XMIT_SUCCESS;
   
10.                 } else {
    
11.                         q->st.pdrop++;
    
12.                 }
    
13.                 kfree_skb(skb);
    
14.                 sch->qstats.drops++;
    
15.                 return NET_XMIT_DROP;
    
16.         }

复制代码

如果队列还没有超限，则入队，重新计算队列长度，计算统计信息。否则，丢弃该包，并做相应统计。

如果大于或等于MAXth：

1.         if (q->qave >= q->qth_max) {
   
2.                 q->qcount = -1;
   
3.                 sch->qstats.overlimits++;     //计算溢出标志
   
4. mark:
   
5.                 //丢包，判断是否有ECN拥塞控制标志
   
6.                                 if  (!(q->flags&TC_RED_ECN) || !red_ecn_mark(skb)) {
   
7.                         q->st.early++;     //标注及丢弃数据包
   
8.                         goto drop;
   
9.                 }
   
10.                 q->st.marked++;
    
11.                 goto enqueue;
    
12.         }

复制代码

如果在两个门限值之间：

1.         if (++q->qcount) {
   
2.                 /* The formula used below causes questions.
   
3. 
   
4.                    OK. qR is random number in the interval 0..Rmask
   
5.                    i.e. 0..(2^Plog). If we used floating point
   
6.                    arithmetics, it would be: (2^Plog)*rnd_num,
   
7.                    where rnd_num is less 1.
   
8. 
   
9.                    Taking into account, that qave have fixed
   
10.                    point at Wlog, and Plog is related to max_P by
    
11.                    max_P = (qth_max-qth_min)/2^Plog; two lines
    
12.                    below have the following floating point equivalent:
    
13.                   
    
14.                    max_P*(qave - qth_min)/(qth_max-qth_min) < rnd/qcount
    
15. 
    
16.                    Any questions? --ANK (980924)
    
17.                  */
    
18.                 if (((q->qave - q->qth_min)>>q->Wlog)*q->qcount < q->qR)
    
19.                         goto enqueue;
    
20.                 q->qcount = 0;
    
21.                 q->qR = net_random()&q->Rmask;
    
22.                 sch->qstats.overlimits++;
    
23.                 goto mark;
    
24.         }

复制代码

开始随机丢包，未丢弃的包通过goto enqueue入队，其它的就丢弃。

这段代码的关键是

1. if (((q->qave - q->qth_min)>>q->Wlog)*q->qcount < q->qR)

复制代码

换言之，就是根据Pb（随机丢包概率）和Pa(一个临时根据)的关系来决定。其中：
Pb = max_P*(qave - qth_min)/(qth_max-qth_min)
Pa = rnd/qcount

因为
max_P = (qth_max-qth_min)/2^Plog;
所以可得：
Pb = ((q->qave - q->qth_min)>>q->Wlog)

上式就可以简化成为：
Pb * qcount < rnd
其中，qcount即为q->qcount，rnd即为q->qR。
而q->qR的计算公式为：
q->qR = net_random()&q->Rmask;
其中掩码位的计算，在初始化中已经进行：
q->Rmask = ctl->Plog < 32 ? ((1<<ctl->Plog) - 1) : ~0UL;

丢包，是通过goto mark跳转来执行的，不过它并不是直接丢弃，而是要先判断数据包是否有没用ECN拥塞标志，如果没有，则直接丢弃。

当平均队列长度在两个阀值之间，而count又为-1时，程序会执行到这一步，重新计算qR，并将数据包入队(这一步还没有怎么看明白^o^)：
        q->qR = net_random()&q->Rmask;
        goto enqueue;

丢包，直接丢弃，统计字段累加。
drop:
        kfree_skb(skb);
        sch->qstats.drops++;
        return NET_XMIT_CN;

基本上就是这些了。

[ 本帖最后由 独孤九贱 于 2008-12-3 16:54 编辑 ]

剑次狼

收藏下，慢慢看

kns1024wh

这个应该放到内核源码里面，什么时候需要定制内核，什么时候好好的研究研究

emmoblin

呵呵，我也在做qos，前段时间一直在研究htb源码。
这个red早晚也得用，收藏之，谢谢lz

Send_linux

:wink:
兄弟，我收藏到本期杂志中去了，不介意吧？

深蓝苹果

好东西啊，学习了