Linux内核中流量控制(14)

本文档的Copyleft归yfydz所有，使用GPL发布，可以自由拷贝，转载，转载时请保持文档的完整性，严禁用于任何商业用途。
msn: [e-mail]
来源：http://yfydz.cublog.cn

5.14  CBQ(Class Based Queueing, 基于类别的排队)

CBQ(Class-Based Queueing discipline)是个经典的基于分类的流控算法，比较复杂, 具体算法基础也是TBF, 也是根据令牌情况决定是否发送数据包。CBQ在HTB出现后一般用得就少了, 其实对于分类的类别操作和HTB也差不多, 所以只简要介绍一下入队和出队操作过程, 其他处理过程不再详细分析了。

5.14.1 CBQ相关结构定义

// CBQ流控结构
static struct Qdisc_ops cbq_qdisc_ops = {
 .next  = NULL,
 .cl_ops  = &cbq_class_ops,
 .id  = "cbq",
// 私有数据是一个用于CBQ调度的数据结构struct cbq_sched_data
 .priv_size = sizeof(struct cbq_sched_data),
 .enqueue = cbq_enqueue,
 .dequeue = cbq_dequeue,
 .requeue = cbq_requeue,
 .drop  = cbq_drop,
 .init  = cbq_init,
 .reset  = cbq_reset,
 .destroy = cbq_destroy,
 .change  = NULL,
 .dump  = cbq_dump,
 .dump_stats = cbq_dump_stats,
 .owner  = THIS_MODULE,
};

// CBQ类别操作结构
static struct Qdisc_class_ops cbq_class_ops = {
 .graft  = cbq_graft,
 .leaf  = cbq_leaf,
 .get  = cbq_get,
 .put  = cbq_put,
 .change  = cbq_change_class,
 .delete  = cbq_delete,
 .walk  = cbq_walk,
 .tcf_chain = cbq_find_tcf,
 .bind_tcf = cbq_bind_filter,
 .unbind_tcf = cbq_unbind_filter,
 .dump  = cbq_dump_class,
 .dump_stats = cbq_dump_class_stats,
};

// CBQ调度数据, 是作为CBQ调度算法的Qdisc的私有数据
struct cbq_sched_data
{
// 类别HASH表, 16个链表
 struct cbq_class *classes[16];  /* Hash table of all classes */
 int   nclasses[TC_CBQ_MAXPRIO+1];
 unsigned  quanta[TC_CBQ_MAXPRIO+1];
 struct cbq_class link;
 unsigned  activemask;
// 活动的CBQ类别链表, 最多8+1个, 按类别优先权区分
 struct cbq_class *active[TC_CBQ_MAXPRIO+1]; /* List of all classes
           with backlog */
#ifdef CONFIG_NET_CLS_POLICE
// 接收用的CBQ类别结构
 struct cbq_class *rx_class;
#endif
// 发送用的CBQ类别结构
 struct cbq_class *tx_class;
// 借带宽时的CBQ类别结构
 struct cbq_class *tx_borrowed;
 int   tx_len;
 psched_time_t  now;  /* Cached timestamp */
 psched_time_t  now_rt;  /* Cached real time */
 unsigned  pmask;
 struct timer_list delay_timer;
 struct timer_list wd_timer; /* Watchdog timer,
         started when CBQ has
         backlog, but cannot
         transmit just now */
 long   wd_expires;
 int   toplevel;
 u32   hgenerator;
};

// CBQ类别结构
struct cbq_class
{
// 链表下一项
 struct cbq_class *next;  /* hash table link */
// backlog中的下一个活动的类别结构
 struct cbq_class *next_alive; /* next class with backlog in this priority band */
/* Parameters */
// 类型ID值
 u32   classid;
// 正常情况下的优先权值
 unsigned char  priority; /* class priority */
// 流量超过限制值后使用的优先权值
 unsigned char  priority2; /* priority to be used after overlimit */
//
 unsigned char  ewma_log; /* time constant for idle time calculation */
// ovl策略
 unsigned char  ovl_strategy;
#ifdef CONFIG_NET_CLS_POLICE
 unsigned char  police;
#endif
// 缺省
 u32   defmap;
 /* Link-sharing scheduler parameters */
 long   maxidle; /* Class parameters: see below. */
 long   offtime;
 long   minidle;
 u32   avpkt;
// 流量表指针
 struct qdisc_rate_table *R_tab;
 /* Overlimit strategy parameters */
// 超过限制时的处理函数
 void   (*overlimit)(struct cbq_class *cl);
 long   penalty;
 /* General scheduler (WRR) parameters */
 long   allot;
 long   quantum; /* Allotment per WRR round */
 long   weight;  /* Relative allotment: see below */
// 指向当前Qdisc结构
 struct Qdisc  *qdisc;  /* Ptr to CBQ discipline */
 struct cbq_class *split;  /* Ptr to split node */
 struct cbq_class *share;  /* Ptr to LS parent in the class tree */
 struct cbq_class *tparent; /* Ptr to tree parent in the class tree */
 struct cbq_class *borrow; /* NULL if class is bandwidth limited;
         parent otherwise */
 struct cbq_class *sibling; /* Sibling chain */
 struct cbq_class *children; /* Pointer to children chain */
// 内部流控节点, 实际通过该Qdisc完成流控操作
 struct Qdisc  *q;  /* Elementary queueing discipline */

/* Variables */
 unsigned char  cpriority; /* Effective priority */
 unsigned char  delayed;
 unsigned char  level;  /* level of the class in hierarchy:
         0 for leaf classes, and maximal
         level of children + 1 for nodes.
       */
 psched_time_t  last;  /* Last end of service */
 psched_time_t  undertime;
 long   avgidle;
 long   deficit; /* Saved deficit for WRR */
 unsigned long  penalized;
 struct gnet_stats_basic bstats;
 struct gnet_stats_queue qstats;
 struct gnet_stats_rate_est rate_est;
 spinlock_t  *stats_lock;
 struct tc_cbq_xstats xstats;
 struct tcf_proto *filter_list;
 int   refcnt;
 int   filters;
 struct cbq_class  *defaults[TC_PRIO_MAX+1];
};

5.14.2 入队操作

// 同样, 对于入队成功的类别节点, 也进行激活操作, 将该类别插入活动类别节点链表,
// 但没使用有序的RB树, 只是普通链表
static int
cbq_enqueue(struct sk_buff *skb, struct Qdisc *sch)
{
// 获取CBQ私有数据
 struct cbq_sched_data *q = qdisc_priv(sch);
 int len = skb->len;
 int ret;
// 对skb数据包进行分类,找到相关类别结构
 struct cbq_class *cl = cbq_classify(skb, sch, &ret);
#ifdef CONFIG_NET_CLS_POLICE
// 如果内核定义了NET_CLS_POLICE, 将该类别结构作为CBQ的数据输入类
 q->rx_class = cl;
#endif
// 如果没找到类别, 丢弃该数据包
 if (cl == NULL) {
  if (ret == NET_XMIT_BYPASS)
   sch->qstats.drops++;
  kfree_skb(skb);
  return ret;
 }
#ifdef CONFIG_NET_CLS_POLICE
// 如果内核定义了NET_CLS_POLICE, 将内部流控节点的父节点指向当前Qdisc
 cl->q->__parent = sch;
#endif
// 调用基本Qdisc的入队函数进行实际入队操作
 if ((ret = cl->q->enqueue(skb, cl->q)) == NET_XMIT_SUCCESS) {
// 操作成功, 进行统计量更新
  sch->q.qlen++;
  sch->bstats.packets++;
  sch->bstats.bytes+=len;
// 修正cbq_sched_data结构中的时间参数now和toplevel参数
  cbq_mark_toplevel(q, cl);
// 如果backlog已经没有排队的类别处理结构, 激活该类别结构
  if (!cl->next_alive)
   cbq_activate_class(cl);
  return ret;
 }
// 入队失败, 丢包, 统计
 sch->qstats.drops++;
 cbq_mark_toplevel(q, cl);
 cl->qstats.drops++;
 return ret;
}

/*
   A packet has just been enqueued on the empty class.
   cbq_activate_class adds it to the tail of active class list
   of its priority band.
 */
// CBQ类别激活, 将该类别结构添加到合适优先权的活动类别链表中
static __inline__ void cbq_activate_class(struct cbq_class *cl)
{
// CBQ私有数据
 struct cbq_sched_data *q = qdisc_priv(cl->qdisc);
// 类别优先权
 int prio = cl->cpriority;
 struct cbq_class *cl_tail;
// 从数据结构看, 是将该类别节点添加到该优先权的活动节点链表的头的位置
 cl_tail = q->active[prio];
 q->active[prio] = cl;
 if (cl_tail != NULL) {
// 原来的链表非空
  cl->next_alive = cl_tail->next_alive;
  cl_tail->next_alive = cl;
 } else {
// 原来的链表空, 设置activemask相应位为1表示该优先权的活动链表可用了
  cl->next_alive = cl;
  q->activemask |= (1<<prio);
 }
}
 
5.14.3 出队操作
 
static struct sk_buff *
cbq_dequeue(struct Qdisc *sch)
{
 struct sk_buff *skb;
// CBQ私有数据
 struct cbq_sched_data *q = qdisc_priv(sch);
 psched_time_t now;
 psched_tdiff_t incr;

// 获取当前时间
 PSCHED_GET_TIME(now);
// 当前时间和结构中记录当前时间的时间差
 incr = PSCHED_TDIFF(now, q->now_rt);

 if (q->tx_class) {
// 发出数据的CBQ类别结构非空, 继续修改时间差
  psched_tdiff_t incr2;
  /* Time integrator. We calculate EOS time
     by adding expected packet transmission time.
     If real time is greater, we warp artificial clock,
     so that:
     cbq_time = max(real_time, work);
   */
// 根据当前发送队列长度计算令牌值作为时间差值2
  incr2 = L2T(&q->link, q->tx_len);
// CBQ当前时间增加
  PSCHED_TADD(q->now, incr2);
// 更新CBQ数据
  cbq_update(q);
// 在当前时间差中减去该差值
  if ((incr -= incr2) < 0)
   incr = 0;
 }
// 增加CBQ当前时间
 PSCHED_TADD(q->now, incr);
// 设置CBQ当前实时时间
 q->now_rt = now;

// 进行循环
 for (;;) {
// wd_expires参数清零
  q->wd_expires = 0;
// 从队列中取一个数据包
  skb = cbq_dequeue_1(sch);
// 如果该数据包存在, 返回, 同时减少数据队列长度
  if (skb) {
   sch->q.qlen--;
   sch->flags &= ~TCQ_F_THROTTLED;
   return skb;
  }
// 以下是没取得数据包时的情况处理
  /* All the classes are overlimit.
     It is possible, if:
     1. Scheduler is empty.
     2. Toplevel cutoff inhibited borrowing.
     3. Root class is overlimit.
     Reset 2d and 3d conditions and retry.
     Note, that NS and cbq-2.0 are buggy, peeking
     an arbitrary class is appropriate for ancestor-only
     sharing, but not for toplevel algorithm.
     Our version is better, but slower, because it requires
     two passes, but it is unavoidable with top-level sharing.
  */
// 如果已经到了分类树的顶层, 而且设置q->link.undertime
  if (q->toplevel == TC_CBQ_MAXLEVEL &&
      PSCHED_IS_PASTPERFECT(q->link.undertime))
   break;
// 设置为顶层, 设置q->link.undertime, 如果这两个参数在下次循环的
// cbq_dequeue_1()函数处理过程中时没改变的话, 循环将终止
  q->toplevel = TC_CBQ_MAXLEVEL;
  PSCHED_SET_PASTPERFECT(q->link.undertime);
 }
 /* No packets in scheduler or nobody wants to give them to us :-(
    Sigh... start watchdog timer in the last case. */
// 运行到这里时还是没取到数据包
// 如果队列长度非空, 可又取不到数据包, 属于阻塞状态了
 if (sch->q.qlen) {
  sch->qstats.overlimits++;
  if (q->wd_expires) {
// 如果需要延迟, 计算延迟时间, 最小一个jiffie, 修改定时器的定时
   long delay = PSCHED_US2JIFFIE(q->wd_expires);
   if (delay <= 0)
    delay = 1;
   mod_timer(&q->wd_timer, jiffies + delay);
   sch->flags |= TCQ_F_THROTTLED;
  }
 }
 return NULL;
}

// 从CBQ队列中去一个数据包
static __inline__ struct sk_buff *
cbq_dequeue_1(struct Qdisc *sch)
{
// CBQ私有数据结构
 struct cbq_sched_data *q = qdisc_priv(sch);
 struct sk_buff *skb;
// 活动链表的优先权掩码, 某位为1表示该位对应的优先权活动链表可用
 unsigned activemask;
// 取活动掩码的后8位, 所以最多是8个优先级别
 activemask = q->activemask&0xFF;

// 循环所有的非0位的活动掩码
 while (activemask) {
// ffz: find first zero in word
// 也就是找activemask中第一个非0位, 作为优先权值, 都是值越小优先权越高, 越先取数据
  int prio = ffz(~activemask);
// 将当前活动掩码第一个非0位置0
  activemask &= ~(1<<prio);
// 根据优先权值从CBQ流控结构中取数据包, 找到则返回
  skb = cbq_dequeue_prio(sch, prio);
  if (skb)
   return skb;
 }
// 循环完所有活动链表还是找不到数据包, 返回空
 return NULL;
}

// 从指定优先权的活动数据包链表中取数据包, 这么长一个函数还被声明为inline的
static __inline__ struct sk_buff *
cbq_dequeue_prio(struct Qdisc *sch, int prio)
{
// CBQ私有数据结构
 struct cbq_sched_data *q = qdisc_priv(sch);
 struct cbq_class *cl_tail, *cl_prev, *cl;
 struct sk_buff *skb;
 int deficit;
// prio优先权对应的链表头节点
 cl_tail = cl_prev = q->active[prio];
// 链表头的下一个类别结构作为当前类别结构准备循环整个链表
 cl = cl_prev->next_alive;
// 进入循环
 do {
// 每次循环赤字标志都初始化为0
  deficit = 0;
  /* Start round */
// 链表循环
  do {
// 将当前的类别结构赋值给borrow
   struct cbq_class *borrow = cl;
// 如果该类别的内部流控结构的队列长度非空, 重新查找可borrow的类别,
// 但如果找不到可borrow的类别节点, 跳转
   if (cl->q->q.qlen &&
       (borrow = cbq_under_limit(cl)) == NULL)
    goto skip_class;
// 当前类别赤字不超过0了, 增加一个定额, 同时当前处理类别跳到下一个类别
   if (cl->deficit <= 0) {
    /* Class exhausted its allotment per
       this round. Switch to the next one.
     */
// 赤字标志为1, 需要继续循环
    deficit = 1;
    cl->deficit += cl->quantum;
    goto next_class;
   }
// 调用内部流控节点的出队函数
   skb = cl->q->dequeue(cl->q);
   /* Class did not give us any skb :-(
      It could occur even if cl->q->q.qlen != 0
      f.e. if cl->q == "tbf"
    */
// 从该类别节点没能获取数据包, 跳过该类别
   if (skb == NULL)
    goto skip_class;
// 以下是取数据包成功的情况
// 类别结构的赤字减少数据包长度的量
   cl->deficit -= skb->len;
// 设置发送的CBQ类类别为当前类别结构
   q->tx_class = cl;
// 设置发送borrowed为当前borrow
   q->tx_borrowed = borrow;
// 如果borrow和cl不同, 是调用了cbq_under_limit()重新获取borrow, 此时重新统计数据
   if (borrow != cl) {
#ifndef CBQ_XSTATS_BORROWS_BYTES
    borrow->xstats.borrows++;
    cl->xstats.borrows++;
#else
    borrow->xstats.borrows += skb->len;
    cl->xstats.borrows += skb->len;
#endif
   }
// 流控的发送长度就是数据包长度
   q->tx_len = skb->len;
// 如果当前类别赤字不超过0
   if (cl->deficit <= 0) {
// 该优先权的活动类别链表头设置为该类别
    q->active[prio] = cl;
// cl更新到下一个活动的类别节点
    cl = cl->next_alive;
// cl增加一个定额, 问题是为什么不是原来的cl增加定额而是新的这个cl增加定额
    cl->deficit += cl->quantum;
   }
// 返回数据包
   return skb;
skip_class:
// 以下是取数据包失败的情况
// 如果该类别内部流控队列长度为0或者是被惩罚的(优先权值改变了)
   if (cl->q->q.qlen == 0 || prio != cl->cpriority) {
    /* Class is empty or penalized.
       Unlink it from active chain.
     */
// 该优先权的活动链表头更新, cl从该活动链表中断开
    cl_prev->next_alive = cl->next_alive;
    cl->next_alive = NULL;
    /* Did cl_tail point to it? */
    if (cl == cl_tail) {
// 原来那个节点是链表末尾节点的情况, 删除后更新末尾节点的指针
     /* Repair it! */
     cl_tail = cl_prev;
     /* Was it the last class in this band? */
     if (cl == cl_tail) {
// 整个链表都空了的情况, 将对应优先权标志清除, 该优先权活动队列已经不可用
      /* Kill the band! */
      q->active[prio] = NULL;
      q->activemask &= ~(1<<prio);
// 如果该类别中还有数据, 重新分配到合适的活动链表
      if (cl->q->q.qlen)
       cbq_activate_class(cl);
// 返回空数据包
      return NULL;
     }
// 更新CBQ流控中记录该链表的头指针为末尾节点指针
     q->active[prio] = cl_tail;
    }
// 如果该类别中还有数据, 重新分配到合适的活动链表
    if (cl->q->q.qlen)
     cbq_activate_class(cl);
// 更新当前cl为保存的前一个节点, 也就是要跳过该节点
    cl = cl_prev;
   }
next_class:
// 保存当前cl到cl_prev
   cl_prev = cl;
// 更新cl位下一个活动cl节点
   cl = cl->next_alive;
// 循环链表所有节点
  } while (cl_prev != cl_tail);
// 赤字非0就一直循环
 } while (deficit);
// 一直没取到数据包, 更新CBQ流控中记录该优先权链表的头指针, 返回空
// 这时的cl_prev应该等于cl_tail
 q->active[prio] = cl_prev;
 return NULL;
}

// 查找可借带宽的类别节点, 返回空表示查找失败
static __inline__ struct cbq_class *
cbq_under_limit(struct cbq_class *cl)
{
 struct cbq_sched_data *q = qdisc_priv(cl->qdisc);
 struct cbq_class *this_cl = cl;
// 父节点为空, 是根节点, 返回类别自身
 if (cl->tparent == NULL)
  return cl;
 if (PSCHED_IS_PASTPERFECT(cl->undertime) ||
     !PSCHED_TLESS(q->now, cl->undertime)) {
// 如果undertime为0或当前流控时间已经过了undertime, 返回类别本身
  cl->delayed = 0;
  return cl;
 }
 do {
  /* It is very suspicious place. Now overlimit
     action is generated for not bounded classes
     only if link is completely congested.
     Though it is in agree with ancestor-only paradigm,
     it looks very stupid. Particularly,
     it means that this chunk of code will either
     never be called or result in strong amplification
     of burstiness. Dangerous, silly, and, however,
     no another solution exists.
   */
// 类别更新为可借带宽的节点
  if ((cl = cl->borrow) == NULL) {
// 如果是空的,表示不可借, 阻塞了, 返回空
   this_cl->qstats.overlimits++;
   this_cl->overlimit(this_cl);
   return NULL;
  }
// 如果该类别层次过大, 返回空
  if (cl->level > q->toplevel)
   return NULL;
// 如果新找到的类别cl的undertime非0而且当前流控时间还没到undertime, 就一直循环
// 因为有上面两个返回条件, 循环次数是不会太多的
 } while (!PSCHED_IS_PASTPERFECT(cl->undertime) &&
   PSCHED_TLESS(q->now, cl->undertime));
// 延迟标志置0, 返回找到的可borrow的类别
 cl->delayed = 0;
 return cl;
}
 
 
...... 待续 ......