代码timeafter=3500什么意思_系统运维

博客园首页联系管理

jiffies相关时间比较函数time_after、time_before详解

1. jiffies简介

首先， *** 作系统有个系统专用定时器(system timer)，俗称滴答定时器，或者系统心跳。

全局变量jiffies取值为自 *** 作系统启动以来的时钟滴答的数目，数据类型为 unsigned long volatile (32位无符号长整型)，最大取值是2^32-1。

2. jiffies与秒的转换

将 jiffies转换为秒，可采用公式：(jiffies/HZ) 计算。

将秒转换为jiffies，可采用公式：(seconds*HZ) 计算。

示例(本博客后面将介绍涉及到的time_before)：

unsgned long delay = jiffies + 2*HZ

while(time_before(jiffies, delay))// 忙等待两秒，占用CPU的一个核心，期间不执行调度

3. jiffies的溢出介绍

当时钟中断发生时，jiffies值就加1。

假定HZ=100，那么1个jiffies等于1/100 秒，jiffies可记录的最大秒数为(2^32 -1)/100=42949672.95秒，约合497天或1.38年，

当取值到达最大值时仍继续加1，就变为了0！

即HZ=100时，连续累加的溢出时间是一年又四个多月，如果程序对jiffies的溢出没有加以充分考虑，那么在连续运行一年又四个多月后，这些程序还能够稳定运行吗？

4. 示例1，一个 jiffies溢出造成程序逻辑出错的示例

复制代码

unsigned long timeout = jiffies + HZ/2/* timeout in 0.5s */

/* do some work ... */

do_somework()

/* then see whether we took too long */

if (timeout >jiffies) {

/* we did not time out, call no_timeout_handler() ... */

no_timeout_handler()

} else {

/* we timed out, call timeout_handler() ... */

timeout_handler()

}

复制代码

本例的意图：

从当前时间起，如果在0.5秒内执行完do_somework()，则调用no_timeout_handler()。如果在0.5秒后执行完do_somework()，则调用timeout_handler()。

然后当溢出时呢？该意图会被打破吗？

假设程序开始执行前，timeout值已经接近最大值(即2^32-1 ) ，jiffies的值是(timeout-HZ/2)，

之后do_some_work执行了挺久(超过0.5秒)，jiffies的值也发生了溢出(jiffies做自增 *** 作的中途超过了32位无符号数的最大值)，

溢出后的值，可能是很小的一个数字，所以造成jiffies的值 <timeout,

之后的代码执行流就走到了no_time_handler()这里，这显然和程序设计者的初衷(意图)是违背的。

5. Linux内核如何来防止jiffies溢出

Linux内核中提供了一些宏，可有效地解决由于jiffies溢出而造成程序逻辑出错的情况。

PS：下图源自Linux Kernel version 3.10.14

* time_after：

* time_after(a,b) returns true if the time a is after time b.

同时根据 #define time_before(a,b)time_after(b,a) ，我们可以知道

* time_before(a,b) returns true if the time b is after time a.

6. time_after 在驱动代码中的应用展示

7. time_after等用于时间比较的宏的原理简介

下面的文字摘录自博文：https://blog.csdn.net/jk110333/article/details/8177285

读者先大致浏览一遍即可，不必纠结于绞尽脑汁的细节理解, 后面我将表达个人理解，读者也可以直接向下浏览，看我的个人理解。

/**********************************开始摘录********************************************/

　我们仍然以8位无符号整型(unsigned char)为例来加以说明。仿照上面的time_after宏，我们可以给出简化的8位无符号整型对应的after宏：

　#define uc_after(a, b) ((char)(b) - (char)(a) <0)

设a和b的数据类型为unsigned char，b为临近8位无符号整型最大值附近的一个固定值254，下面给出随着a（设其初始值为254）变化而得到的计算值：

a b (char)(b) - (char)(a)

254 254 0

255 - 1

0 - 2

1 - 3

...

124 -126

125 -127

126 -128

127 127

128 126

...

252 2

253 1

　从上面的计算可以看出，设定b不变，随着a（设其初始值为254）不断增长1，a的取值变化为：

254, 255, (一次产生溢出)

0, 1, ..., 124, 125, 126, 127, 126, ..., 253, 254, 255, (二次产生溢出)

0, 1, ...

...

而(char)(b) - (char)(a)的变化为：

0, -1,

-2, -3, ..., -126, -127, -128, 127, 126, ..., 1, 0, -1,

-2, -3, ...

...

从上面的详细过程可以看出，当a取值为254，255, 接着在（一次产生溢出）之后变为0，然后增长到127之前，uc_after(a,b)的结果都显示a是在b之后，这也与我们的预期相符。但在a取值为 127之后， uc_after(a,b)的结果却显示a是在b之前。

从上面的运算过程可以得出以下结论：

使用uc_after(a,b)宏来计算两个8位无符号整型a和b之间的大小（或先/后，before/after），那么a和b的取值应当满足以下限定条件：

. 两个值之间相差从逻辑值来讲应小于有符号整型的最大值。

. 对于8位无符号整型，两个值之间相差从逻辑值来讲应小于128。

从上面可以类推出以下结论：

对于time_after等比较jiffies先/后的宏，两个值的取值应当满足以下限定条件：

两个值之间相差从逻辑值来讲应小于有符号整型的最大值。

对于32位无符号整型，两个值之间相差从逻辑值来讲应小于2147483647。

对于HZ=100，那么两个时间值之间相差不应当超过2147483647/100秒 = 0.69年 = 248.5天。

对于HZ=60，那么两个时间值之间相差不应当超过2147483647/60秒 = 1.135年。

在实际代码应用中，需要比较的先/后的两个时间值之间一般都相差很小，范围大致在1秒~1天左右，所以以上time_after等比较时间先 /后的宏完全可以放心地用于实际的代码中。　

/***********************************摘录结束******************************************/

看完这段文字，感觉有点绕的，那么原理到底是啥呢？是一堆数学计算吗？是啊，就是这数学规律！

凡事都是有利有弊的，针对一件事物的优化，有利处，必然带来不利之处，从哲学角度来进行理解，事物的两面性。

本文第4部分，示例1介绍了jiffies的一个例子，它的弊处是会溢出，如果我们不抓住溢出这个弊处来看待这件事物，那么timeout的值可以做的很大，这是优势。

然而溢出是真实存在的，无法满足客观需求的，所以需要改进，

从该数学规律入手进行改进后，不溢出了，这是优势，

但是改进后对timeout的值也缩小了使用范围，这是为了达到该优势所带来的必要开销或损耗。这就是事物的两面性。

8. 示例2，对示例1进行改进：使用time_before宏后的正确代码

复制代码

unsigned long timeout = jiffies + HZ/2/* timeout in 0.5s */

/* do some work ... */

do_somework()

/* then see whether we took too long */

if (time_before(jiffies, timeout)) {

/* we did not time out, call no_timeout_handler() ... */

no_timeout_handler()

} else {

/* we timed out, call timeout_handler() ... */

timeout_handler()

}

复制代码

/************* 社会的有色眼光是：博士生、研究生、本科生、车间工人重点大学高材生、普通院校、二流院校、野鸡大学年薪百万、五十万、五万这些都只是帽子,可以失败千百次，但我和社会都觉得，人只要成功一次，就能换一顶帽子，只是社会看不见你之前的失败的帽子。当然，换帽子决不是最终目的，走好自己的路就行。杭州.大话西游 *******/

分类: Linux驱动

标签: 内核编程

好文要顶关注我收藏该文

一匹夫

粉丝 - 28 关注 - 3

+加关注

« 上一篇：为什么我觉得需要熟悉vim使用，难道仅仅是为了耍酷？

» 下一篇：九鼎S5PV210开发板的SD卡启动、uboot tftp升级内核镜像

posted @ 2021-01-30 14:39 一匹夫阅读(2508) 评论(0) 编辑收藏举报

刷新评论刷新页面返回顶部

登录后才能查看或发表评论，立即登录或者逛逛博客园首页

【推荐】阿里云新人特惠，爆款云服务器2核4G低至0.46元/天

编辑推荐：

· .Net 6 使用 Consul 实现服务注册与发现

· SQLSERVER 的复合索引和包含索引到底有啥区别？

· [ASP.NET Core] 按用户等级授权

· 深入理解 Linux 物理内存分配全链路实现

· 巧用视觉障眼法，还原 3D 文字特效

阅读排行：

· 既然有MySQL了，为什么还要有MongoDB？

· C#开发的插件程序 - 开源研究系列文章

· 2022年工作总结，迟到比没到好

· 20 张图带你全面了解 HTTPS 协议，再也不怕面试问到了！

· .net core *** 作MongoDB

公告

音乐2 - 林海

00:00 / 00:00An audio error has occurred, player will skip forward in 2 seconds.

1 音乐1Valentin

2 音乐2林海

3 音乐3赵海洋

昵称：一匹夫

园龄： 5年9个月

粉丝： 28

关注： 3

+加关注

< 2023年1月 >

日一二三四五六

1 2 3 4 5 6 7

8 9 10 11 12 13 14

15 16 17 18 19 20 21

22 23 24 25 26 27 28

29 30 31 1 2 3 4

5 6 7 8 9 10 11

搜索

找找看

谷歌搜索

常用链接

我的随笔

我的评论

我的参与

最新评论

1. Re:在KEIL下查看单片机编程内存使用情况

@HQ_嗨海谢谢...

--一匹夫

2. Re:如何更好地谋生，从事嵌入式软件开发五年的感悟和职业焦虑

说的不错

--Chance_21_12_12

3. Re:在KEIL下查看单片机编程内存使用情况

感谢大佬

--HQ_嗨海

天重点对linux网络数据包的处理做下分析，但是并不关系到上层协议，仅仅到链路层。之前转载过一篇文章，对NAPI做了比较详尽的分析，本文结合Linux内核源代码，对当前网络数据包的处理进行梳理。根据NAPI的处理特性，对设备提出一定的要求1、设备需要有足够的缓冲区，保存多个数据分组2、可以禁用当前设备中断，然而不影响其他的 *** 作。当前大部分的设备都支持NAPI，但是为了对之前的保持兼容，内核还是对之前中断方式提供了兼容。我们先看下NAPI具体的处理方式。我们都知道中断分为中断上半部和下半部，上半部完成的任务很是简单，仅仅负责把数据保存下来；而下半部负责具体的处理。为了处理下半部，每个CPU有维护一个softnet_data结构。我们不对此结构做详细介绍，仅仅描述和NAPI相关的部分。结构中有一个poll_list字段，连接所有的轮询设备。还维护了两个队列input_pkt_queue和process_queue。这两个用户传统不支持NAPI方式的处理。前者由中断上半部的处理函数吧数据包入队，在具体的处理时，使用后者做中转，相当于前者负责接收，后者负责处理。最后是一个napi_struct的backlog，代表一个虚拟设备供轮询使用。在支持NAPI的设备下，每个设备具备一个缓冲队列，存放到来数据。每个设备对应一个napi_struct结构，该结构代表该设备存放在poll_list中被轮询。而设备还需要提供一个poll函数，在设备被轮询到后，会调用poll函数对数据进行处理。基本逻辑就是这样，下面看下具体流程。中断上半部：非NAPI：非NAPI对应的上半部函数为netif_rx，位于Dev.,c中int netif_rx(struct sk_buff *skb){int ret/* if netpoll wants it, pretend we never saw it *//*如果是net_poll想要的，则不作处理*/if (netpoll_rx(skb))return NET_RX_DROP/*检查时间戳*/net_timestamp_check(netdev_tstamp_prequeue, skb)trace_netif_rx(skb)#ifdef CONFIG_RPSif (static_key_false(&rps_needed)) {struct rps_dev_flow voidflow, *rflow = &voidflowint cpu/*禁用抢占*/preempt_disable()rcu_read_lock()cpu = get_rps_cpu(skb->dev, skb, &rflow)if (cpu <0)cpu = smp_processor_id()/*把数据入队*/ret = enqueue_to_backlog(skb, cpu, &rflow->last_qtail)rcu_read_unlock()preempt_enable()} else#endif{ unsigned int qtailret = enqueue_to_backlog(skb, get_cpu(), &qtail)put_cpu()}return ret}中间RPS暂时不关心，这里直接调用enqueue_to_backlog放入CPU的全局队列input_pkt_queuestatic int enqueue_to_backlog(struct sk_buff *skb, int cpu,unsigned int *qtail){struct softnet_data *sdunsigned long flags/*获取cpu相关的softnet_data变量*/sd = &per_cpu(softnet_data, cpu)/*关中断*/local_irq_save(flags)rps_lock(sd)/*如果input_pkt_queue的长度小于最大限制，则符合条件*/if (skb_queue_len(&sd->input_pkt_queue) <= netdev_max_backlog) {/*如果input_pkt_queue不为空，说明虚拟设备已经得到调度，此时仅仅把数据加入input_pkt_queue队列即可*/if (skb_queue_len(&sd->input_pkt_queue)) {enqueue:__skb_queue_tail(&sd->input_pkt_queue, skb)input_queue_tail_incr_save(sd, qtail)rps_unlock(sd)local_irq_restore(flags)return NET_RX_SUCCESS}/* Schedule NAPI for backlog device* We can use non atomic operation since we own the queue lock*//*否则需要调度backlog 即虚拟设备，然后再入队。napi_struct结构中的state字段如果标记了NAPI_STATE_SCHED,则表明该设备已经在调度，不需要再次调度*/if (!__test_and_set_bit(NAPI_STATE_SCHED, &sd->backlog.state)) {if (!rps_ipi_queued(sd))____napi_schedule(sd, &sd->backlog)}goto enqueue}/*到这里缓冲区已经不足了，必须丢弃*/sd->dropped++rps_unlock(sd)local_irq_restore(flags)atomic_long_inc(&skb->dev->rx_dropped)kfree_skb(skb)return NET_RX_DROP}该函数逻辑也比较简单，主要注意的是设备必须先添加调度然后才能接受数据，添加调度调用了____napi_schedule函数，该函数把设备对应的napi_struct结构插入到softnet_data的poll_list链表尾部，然后唤醒软中断，这样在下次软中断得到处理时，中断下半部就会得到处理。不妨看下源码static inline void ____napi_schedule(struct softnet_data *sd,struct napi_struct *napi){list_add_tail(&napi->poll_list, &sd->poll_list)__raise_softirq_irqoff(NET_RX_SOFTIRQ)}NAPI方式NAPI的方式相对于非NAPI要简单许多，看下e100网卡的中断处理函数e100_intr，核心部分if (likely(napi_schedule_prep(&nic->napi))) {e100_disable_irq(nic)//屏蔽当前中断__napi_schedule(&nic->napi)//把设备加入到轮训队列}if条件检查当前设备是否可被调度，主要检查两个方面：1、是否已经在调度 2、是否禁止了napi pending.如果符合条件，就关闭当前设备的中断，调用__napi_schedule函数把设备假如到轮训列表，从而开启轮询模式。分析：结合上面两种方式，还是可以发现两种方式的异同。其中softnet_data作为主导结构，在NAPI的处理方式下，主要维护轮询链表。NAPI设备均对应一个napi_struct结构，添加到链表中；非NAPI没有对应的napi_struct结构，为了使用NAPI的处理流程，使用了softnet_data结构中的back_log作为一个虚拟设备添加到轮询链表。同时由于非NAPI设备没有各自的接收队列，所以利用了softnet_data结构的input_pkt_queue作为全局的接收队列。这样就处理而言，可以和NAPI的设备进行兼容。但是还有一个重要区别，在NAPI的方式下，首次数据包的接收使用中断的方式，而后续的数据包就会使用轮询处理了；而非NAPI每次都是通过中断通知。下半部：下半部的处理函数，之前提到，网络数据包的接发对应两个不同的软中断，接收软中断NET_RX_SOFTIRQ的处理函数对应net_rx_actionstatic void net_rx_action(struct softirq_action *h){struct softnet_data *sd = &__get_cpu_var(softnet_data)unsigned long time_limit = jiffies + 2int budget = netdev_budgetvoid *havelocal_irq_disable()/*遍历轮询表*/while (!list_empty(&sd->poll_list)) {struct napi_struct *nint work, weight/* If softirq window is exhuasted then punt.* Allow this to run for 2 jiffies since which will allow* an average latency of 1.5/HZ.*//*如果开支用完了或者时间用完了*/if (unlikely(budget <= 0 || time_after_eq(jiffies, time_limit)))goto softnet_breaklocal_irq_enable()/* Even though interrupts have been re-enabled, this* access is safe because interrupts can only add new* entries to the tail of this list, and only ->poll()* calls can remove this head entry from the list.*//*获取链表中首个设备*/n = list_first_entry(&sd->poll_list, struct napi_struct, poll_list)have = netpoll_poll_lock(n)weight = n->weight/* This NAPI_STATE_SCHED test is for avoiding a race* with netpoll's poll_napi(). Only the entity which* obtains the lock and sees NAPI_STATE_SCHED set will* actually make the ->poll() call. Therefore we avoid* accidentally calling ->poll() when NAPI is not scheduled.*/work = 0/*如果被设备已经被调度，则调用其处理函数poll函数*/if (test_bit(NAPI_STATE_SCHED, &n->state)) {work = n->poll(n, weight)//后面weight指定了一个额度trace_napi_poll(n)}WARN_ON_ONCE(work >weight)/*总额度递减*/budget -= worklocal_irq_disable()/* Drivers must not modify the NAPI state if they* consume the entire weight. In such cases this code* still "owns" the NAPI instance and therefore can* move the instance around on the list at-will.*//*如果work=weight的话。任务就完成了，把设备从轮询链表删除*/if (unlikely(work == weight)) {if (unlikely(napi_disable_pending(n))) {local_irq_enable()napi_complete(n)local_irq_disable()} else {if (n->gro_list) {/* flush too old packets* If HZ <1000, flush all packets.*/local_irq_enable()napi_gro_flush(n, HZ >= 1000)local_irq_disable()}/*每次处理完就把设备移动到列表尾部*/list_move_tail(&n->poll_list, &sd->poll_list)}}netpoll_poll_unlock(have)}out:net_rps_action_and_irq_enable(sd)#ifdef CONFIG_NET_DMA/** There may not be any more sk_buffs coming right now, so push* any pending DMA copies to hardware*/dma_issue_pending_all()#endifreturnsoftnet_break:sd->time_squeeze++__raise_softirq_irqoff(NET_RX_SOFTIRQ)goto out}这里有处理方式比较直观，直接遍历poll_list链表，处理之前设置了两个限制：budget和time_limit。前者限制本次处理数据包的总量，后者限制本次处理总时间。只有二者均有剩余的情况下，才会继续处理。处理期间同样是开中断的，每次总是从链表表头取设备进行处理，如果设备被调度，其实就是检查NAPI_STATE_SCHED位，则调用 napi_struct的poll函数，处理结束如果没有处理完，则把设备移动到链表尾部，否则从链表删除。NAPI设备对应的poll函数会同样会调用__netif_receive_skb函数上传协议栈，这里就不做分析了，感兴趣可以参考e100的poll函数e100_poll。而非NAPI对应poll函数为process_backlog。static int process_backlog(struct napi_struct *napi, int quota){int work = 0struct softnet_data *sd = container_of(napi, struct softnet_data, backlog)#ifdef CONFIG_RPS/* Check if we have pending ipi, its better to send them now,* not waiting net_rx_action() end.*/if (sd->rps_ipi_list) {local_irq_disable()net_rps_action_and_irq_enable(sd)}#endifnapi->weight = weight_plocal_irq_disable()while (work <quota) {struct sk_buff *skbunsigned int qlen/*涉及到两个队列process_queue和input_pkt_queue，数据包到来时首先填充input_pkt_queue，而在处理时从process_queue中取，根据这个逻辑，首次处理process_queue必定为空，检查input_pkt_queue如果input_pkt_queue不为空，则把其中的数据包迁移到process_queue中，然后继续处理，减少锁冲突。*/while ((skb = __skb_dequeue(&sd->process_queue))) {local_irq_enable()/*进入协议栈*/__netif_receive_skb(skb)local_irq_disable()input_queue_head_incr(sd)if (++work >= quota) {local_irq_enable()return work}}rps_lock(sd)qlen = skb_queue_len(&sd->input_pkt_queue)if (qlen)skb_queue_splice_tail_init(&sd->input_pkt_queue,&sd->process_queue)if (qlen <quota - work) {/** Inline a custom version of __napi_complete().* only current cpu owns and manipulates this napi,* and NAPI_STATE_SCHED is the only possible flag set on backlog.* we can use a plain write instead of clear_bit(),* and we dont need an smp_mb() memory barrier.*/list_del(&napi->poll_list)napi->state = 0quota = work + qlen}rps_unlock(sd)}local_irq_enable()return work}函数还是比较简单的，需要注意的每次处理都携带一个配额，即本次只能处理quota个数据包，如果超额了，即使没处理完也要返回，这是为了保证处理器的公平使用。处理在一个while循环中完成，循环条件正是work <quota，首先会从process_queue中取出skb,调用__netif_receive_skb上传给协议栈，然后增加work。当work即将大于quota时，即++work >= quota时，就要返回。当work还有剩余额度，但是process_queue中数据处理完了，就需要检查input_pkt_queue，因为在具体处理期间是开中断的，那么期间就有可能有新的数据包到来，如果input_pkt_queue不为空，则调用skb_queue_splice_tail_init函数把数据包迁移到process_queue。如果剩余额度足够处理完这些数据包，那么就把虚拟设备移除轮询队列。这里有些疑惑就是最后为何要增加额度，剩下的额度已经足够处理这些数据了呀？根据此流程不难发现，其实执行的是在两个队列之间移动数据包，然后再做处理。

欢迎分享，转载请注明来源：内存溢出

原文地址:https://54852.com/yw/7121043.html

代码timeafter=3500什么意思

发表评论

评论列表（0条）