mt7530交换芯片的数据接收中断后，把具体接收数据工作任务、委托到workqueue队列中，让内核work_thread（）线程任务来处理，这部分内容请参考《workqueue工作原理》中的描述。
workqueue基本工作流程框架如下：
框架业务关系：
1，程序把work单加入到workqueue中，就等于把工作安排好，是工单的生成者、派遣者；
2，work_pool是工厂，提供工作的场地，worker是工人，负责执行工单，是消费者；
3，PWQ(pool work queue)是派遣工单给工厂协调者，负责匹配生产者与消费者之间协调；
workqueue框架启动流程：
1，内核kernel_init_freeable初始化函数(in main.c)调用workqueue_init()初始化程序；
2，workqueue_init()函数为每隔CPU创建一个worker_pool池；
3，kthread_create_on_node()创建一个worker进程，并worker_attach_to_pool(work,pool);添加到池子中；
worker执行工作流程:workqueue.c-->static ini worker_thread(void* __worker)
/*
* Finish PREP stage. We're guaranteed to have at least one idle
* worker or that someone else has already assumed the manager
* role. This is where @worker starts participating in concurrency
* management if applicable and concurrency management is restored
* after being rebound. See rebind_workers() for details.
*/
worker_clr_flags(worker, WORKER_PREP | WORKER_REBOUND);
do {
struct work_struct *work =
list_first_entry(&pool->worklist,
struct work_struct, entry);
pool->watchdog_ts = jiffies;
if (likely(!(*work_data_bits(work) & WORK_STRUCT_LINKED))) {
/* optimization path, not strictly necessary */
process_one_work(worker, work); /* worker done work,执行queue中的work任务 */
if (unlikely(!list_empty(&worker->scheduled)))
process_scheduled_works(worker);
} else {
move_linked_works(work, &worker->scheduled, NULL);
process_scheduled_works(worker);
}
} while (keep_working(pool));
数据接收任务单的执行者是谁？
硬件中断后程序最终通过insert_work()函数，把待处理的数据任务放到workqueue中，开篇已经描述worker_pool中已经具有worker线程one-by-one执行工作任务，也及时说数据会被process_one_work(worker, work)程序接收、并送往网络协议栈。
/**
* process_one_work - process single work
* @worker: self
* @work: work to process
*
* Process @work. This function contains all the logics necessary to
* process a single work including synchronization against and
* interaction with other workers on the same cpu, queueing and
* flushing. As long as context requirement is met, any worker can
* call this function to process a work.
*
* CONTEXT:
* spin_lock_irq(pool->lock) which is released and regrabbed.
*/
static void process_one_work(struct worker *worker, struct work_struct *work)
__releases(&pool->lock)
__acquires(&pool->lock)
{
struct pool_workqueue *pwq = get_work_pwq(work);
struct worker_pool *pool = worker->pool;
bool cpu_intensive = pwq->wq->flags & WQ_CPU_INTENSIVE;
int work_color;
struct worker *collision;
#ifdef CONFIG_LOCKDEP
/*
* It is permissible to free the struct work_struct from
* inside the function that is called from it, this we need to
* take into account for lockdep too. To avoid bogus "held
* lock freed" warnings as well as problems when looking into
* work->lockdep_map, make a copy and use that here.
*/
struct lockdep_map lockdep_map;
lockdep_copy_map(&lockdep_map, &work->lockdep_map);
#endif
/* ensure we're on the correct CPU */
WARN_ON_ONCE(!(pool->flags & POOL_DISASSOCIATED) &&
raw_smp_processor_id() != pool->cpu);
/*
* A single work shouldn't be executed concurrently by
* multiple workers on a single cpu. Check whether anyone is
* already processing the work. If so, defer the work to the
* currently executing one.
*/
collision = find_worker_executing_work(pool, work);
if (unlikely(collision)) {
move_linked_works(work, &collision->scheduled, NULL);
return;
}
/* claim and dequeue */
debug_work_deactivate(work);
hash_add(pool->busy_hash, &worker->hentry, (unsigned long)work);
worker->current_work = work;
worker->current_func = work->func;
worker->current_pwq = pwq;
work_color = get_work_color(work);
list_del_init(&work->entry);
/*
* CPU intensive works don't participate in concurrency management.
* They're the scheduler's responsibility. This takes @worker out
* of concurrency management and the next code block will chain
* execution of the pending work items.
*/
if (unlikely(cpu_intensive))
worker_set_flags(worker, WORKER_CPU_INTENSIVE);
/*
* Wake up another worker if necessary. The condition is always
* false for normal per-cpu workers since nr_running would always
* be >= 1 at this point. This is used to chain execution of the
* pending work items for WORKER_NOT_RUNNING workers such as the
* UNBOUND and CPU_INTENSIVE ones.
*/
if (need_more_worker(pool))
wake_up_worker(pool);
/*
* Record the last pool and clear PENDING which should be the last
* update to @work. Also, do this inside @pool->lock so that
* PENDING and queued state changes happen together while IRQ is
* disabled.
*/
set_work_pool_and_clear_pending(work, pool->id);
spin_unlock_irq(&pool->lock);
lock_map_acquire(&pwq->wq->lockdep_map);
lock_map_acquire(&lockdep_map);
/*
* Strictly speaking we should mark the invariant state without holding
* any locks, that is, before these two lock_map_acquire()'s.
*
* However, that would result in:
*
* A(W1)
* WFC(C)
* A(W1)
* C(C)
*
* Which would create W1->C->W1 dependencies, even though there is no
* actual deadlock possible. There are two solutions, using a
* read-recursive acquire on the work(queue) 'locks', but this will then
* hit the lockdep limitation on recursive locks, or simply discard
* these locks.
*
* AFAICT there is no possible deadlock scenario between the
* flush_work() and complete() primitives (except for single-threaded
* workqueues), so hiding them isn't a problem.
*/
lockdep_invariant_state(true);
trace_workqueue_execute_start(work);
worker->current_func(work); /* ！！！ 此回调函数是执行此工作入口，因此需要查看任务工单内容 */
/*
* While we must be careful to not use "work" after this, the trace
* point will only record its address.
*/
trace_workqueue_execute_end(work);
lock_map_release(&lockdep_map);
lock_map_release(&pwq->wq->lockdep_map);
if (unlikely(in_atomic() || lockdep_depth(current) > 0)) {
pr_err("BUG: workqueue leaked lock or atomic: %s/0x%08x/%d\n"
" last function: %pf\n",
current->comm, preempt_count(), task_pid_nr(current),
worker->current_func);
debug_show_held_locks(current);
dump_stack();
}
/*
* The following prevents a kworker from hogging CPU on !PREEMPT
* kernels, where a requeueing work item waiting for something to
* happen could deadlock with stop_machine as such work item could
* indefinitely requeue itself while all other CPUs are trapped in
* stop_machine. At the same time, report a quiescent RCU state so
* the same condition doesn't freeze RCU.
*/
cond_resched_rcu_qs();
spin_lock_irq(&pool->lock);
/* clear cpu intensive status */
if (unlikely(cpu_intensive))
worker_clr_flags(worker, WORKER_CPU_INTENSIVE);
/* we're done with it, release */
hash_del(&worker->hentry);
worker->current_work = NULL;
worker->current_func = NULL;
worker->current_pwq = NULL;
worker->desc_valid = false;
pwq_dec_nr_in_flight(pwq, work_color);
}
数据是如何被接收的？
网卡在被ifup | ifconfig up，会触发网卡驱动的fe_probe函数，函数调用关系如下：
static int fe_probe(struct platform_device *pdev)
{
fe_base = devm_ioremap_resource(&pdev->dev, res);
netdev = alloc_etherdev(sizeof(*priv));
SET_NETDEV_DEV(netdev, &pdev->dev);
netdev->netdev_ops = &fe_netdev_ops;
netdev->base_addr = (unsigned long)fe_base;
netdev->irq = platform_get_irq(pdev, 0);
priv = netdev_priv(netdev);
INIT_WORK(&priv->pending_work, fe_pending_work); /* 初始化 work 工单,worker-> func () 回调函数 */
netif_napi_add(netdev, &priv->rx_napi, fe_poll, napi_weight); /* 注册数据接收函数 fe_poll 函数 */
fe_set_ethtool_ops(netdev);
err = register_netdev(netdev);
}
static void fe_pending_work(struct work_struct *work)
{
struct fe_priv *priv = container_of(work, struct fe_priv, pending_work);
int i;
bool pending;
for (i = 0; i < ARRAY_SIZE(fe_work); i++) {
pending = test_and_clear_bit(fe_work[i].bitnr,
priv->pending_flags);
if (pending)
fe_work[i].action(priv); /* 激活 任务标识 */
}
}
/* 数据接收函数入口 */
static int fe_poll(struct napi_struct *napi, int budget)
{
struct fe_priv *priv = container_of(napi, struct fe_priv, rx_napi);
struct fe_hw_stats *hwstat = priv->hw_stats;
int tx_done, rx_done, tx_again;
u32 status, fe_status, status_reg, mask;
u32 tx_intr, rx_intr, status_intr;
status = fe_reg_r32(FE_REG_FE_INT_STATUS);
fe_status = status;
tx_intr = priv->soc->tx_int;
rx_intr = priv->soc->rx_int;
status_intr = priv->soc->status_int;
tx_done = 0;
rx_done = 0;
tx_again = 0;
if (fe_reg_table[FE_REG_FE_INT_STATUS2]) {
fe_status = fe_reg_r32(FE_REG_FE_INT_STATUS2);
status_reg = FE_REG_FE_INT_STATUS2;
} else {
status_reg = FE_REG_FE_INT_STATUS;
}
if (status & tx_intr)
tx_done = fe_poll_tx(priv, budget, tx_intr, &tx_again); /* 调用数据发送函数 fe_poll_tx() 函数 */
if (status & rx_intr)
rx_done = fe_poll_rx(napi, budget, priv, rx_intr); /* 调用数据接收函数 fe_poll_rx() 函数 */
if (unlikely(fe_status & status_intr)) {
if (hwstat && spin_trylock(&hwstat->stats_lock)) {
fe_stats_update(priv);
spin_unlock(&hwstat->stats_lock);
}
fe_reg_w32(status_intr, status_reg);
}
if (unlikely(netif_msg_intr(priv))) {
mask = fe_reg_r32(FE_REG_FE_INT_ENABLE);
netdev_info(priv->netdev,
"done tx %d, rx %d, intr 0x%08x/0x%x\n",
tx_done, rx_done, status, mask);
}
if (!tx_again && (rx_done < budget)) {
status = fe_reg_r32(FE_REG_FE_INT_STATUS);
if (status & (tx_intr | rx_intr)) {
/* let napi poll again */
rx_done = budget;
goto poll_again;
}
napi_complete_done(napi, rx_done);
fe_int_enable(tx_intr | rx_intr);
} else {
rx_done = budget;
}
poll_again:
return rx_done;
}
/* 数据如何被读取出来、送往内核网络协议栈 */
static int fe_poll_rx(struct napi_struct *napi, int budget,
struct fe_priv *priv, u32 rx_intr)
{
struct net_device *netdev = priv->netdev;
struct net_device_stats *stats = &netdev->stats;
struct fe_soc_data *soc = priv->soc;
struct fe_rx_ring *ring = &priv->rx_ring;
int idx = ring->rx_calc_idx;
u32 checksum_bit;
struct sk_buff *skb;
u8 *data, *new_data;
struct fe_rx_dma *rxd, trxd;
int done = 0, pad;
if (netdev->features & NETIF_F_RXCSUM)
checksum_bit = soc->checksum_bit;
else
checksum_bit = 0;
if (priv->flags & FE_FLAG_RX_2B_OFFSET)
pad = 0;
else
pad = NET_IP_ALIGN;
while (done < budget) {
unsigned int pktlen;
dma_addr_t dma_addr;
/* 环形缓冲区获取数据 */
idx = NEXT_RX_DESP_IDX(idx);
rxd = &ring->rx_dma[idx];
data = ring->rx_data[idx];
fe_get_rxd(&trxd, rxd);
if (!(trxd.rxd2 & RX_DMA_DONE))
break;
/* alloc new buffer */
new_data = page_frag_alloc(&ring->frag_cache, ring->frag_size,
GFP_ATOMIC);
if (unlikely(!new_data)) {
stats->rx_dropped++;
goto release_desc;
}
dma_addr = dma_map_single(&netdev->dev,
new_data + NET_SKB_PAD + pad,
ring->rx_buf_size,
DMA_FROM_DEVICE);
if (unlikely(dma_mapping_error(&netdev->dev, dma_addr))) {
skb_free_frag(new_data);
goto release_desc;
}
/* receive data */
skb = build_skb(data, ring->frag_size);
if (unlikely(!skb)) {
skb_free_frag(new_data);
goto release_desc;
}
skb_reserve(skb, NET_SKB_PAD + NET_IP_ALIGN);
dma_unmap_single(&netdev->dev, trxd.rxd1,
ring->rx_buf_size, DMA_FROM_DEVICE);
pktlen = RX_DMA_GET_PLEN0(trxd.rxd2);
skb->dev = netdev;
skb_put(skb, pktlen);
if (trxd.rxd4 & checksum_bit)
skb->ip_summed = CHECKSUM_UNNECESSARY;
else
skb_checksum_none_assert(skb);
skb->protocol = eth_type_trans(skb, netdev); /* 获取协议号 */
if (netdev->features & NETIF_F_HW_VLAN_CTAG_RX &&
RX_DMA_VID(trxd.rxd3))
__vlan_hwaccel_put_tag(skb, htons(ETH_P_8021Q),
RX_DMA_VID(trxd.rxd3)); /* VLAN 打 tag 标 */
#ifdef CONFIG_NET_MEDIATEK_OFFLOAD /* NET_MEDIATEK_OFFLOAD 网络功能卸载 */
if (mtk_offload_check_rx(priv, skb, trxd.rxd4) == 0) {
#endif
stats->rx_packets++;
stats->rx_bytes += pktlen;
napi_gro_receive(napi, skb); /* 调用 napi_gro_receive() 接收数据*/
#ifdef CONFIG_NET_MEDIATEK_OFFLOAD
} else {
dev_kfree_skb(skb);
}
#endif
ring->rx_data[idx] = new_data;
rxd->rxd1 = (unsigned int)dma_addr;
release_desc:
if (priv->flags & FE_FLAG_RX_SG_DMA)
rxd->rxd2 = RX_DMA_PLEN0(ring->rx_buf_size);
else
rxd->rxd2 = RX_DMA_LSO;
ring->rx_calc_idx = idx;
/* make sure that all changes to the dma ring are flushed before
* we continue
*/
wmb();
fe_reg_w32(ring->rx_calc_idx, FE_REG_RX_CALC_IDX0);
done++;
}
if (done < budget)
fe_reg_w32(rx_intr, FE_REG_FE_INT_STATUS);
return done;
}
gro_result_t napi_gro_receive(struct napi_struct *napi, struct sk_buff *skb)
{
skb_mark_napi_id(skb, napi);
trace_napi_gro_receive_entry(skb);
skb_gro_reset_offset(skb);
return napi_skb_finish(dev_gro_receive(napi, skb), skb);
}
EXPORT_SYMBOL(napi_gro_receive);
static gro_result_t napi_skb_finish(gro_result_t ret, struct sk_buff *skb)
{
switch (ret) {
case GRO_NORMAL:
if (netif_receive_skb_internal(skb)) /* ip层的数据接收 */
ret = GRO_DROP;
break;
case GRO_DROP:
kfree_skb(skb);
break;
case GRO_MERGED_FREE:
if (NAPI_GRO_CB(skb)->free == NAPI_GRO_FREE_STOLEN_HEAD)
napi_skb_free_stolen_head(skb);
else
__kfree_skb(skb);
break;
case GRO_HELD:
case GRO_MERGED:
case GRO_CONSUMED:
break;
}
return ret;
}
static int netif_receive_skb_internal(struct sk_buff *skb)
{
int ret;
net_timestamp_check(netdev_tstamp_prequeue, skb);
if (skb_defer_rx_timestamp(skb))
return NET_RX_SUCCESS;
if (static_key_false(&generic_xdp_needed)) {
int ret;
preempt_disable();
rcu_read_lock();
ret = do_xdp_generic(rcu_dereference(skb->dev->xdp_prog), skb);
rcu_read_unlock();
preempt_enable();
if (ret != XDP_PASS)
return NET_RX_DROP;
}
rcu_read_lock();
#ifdef CONFIG_RPS
if (static_key_false(&rps_needed)) {
struct rps_dev_flow voidflow, *rflow = &voidflow;
int cpu = get_rps_cpu(skb->dev, skb, &rflow);
if (cpu >= 0) {
ret = enqueue_to_backlog(skb, cpu, &rflow->last_qtail);
rcu_read_unlock();
return ret;
}
}
#endif
ret = __netif_receive_skb(skb); /* 调用数据接收函数 __netif_receive_skb() */
rcu_read_unlock();
return ret;
}
__netif_receive_skb() --> __netif_receive_skb_core() --> deliver_skb() 函数调用关系，最终调用 协议栈分发函数 deliver_skb() ；
/* skb 发送到网络协议栈 */
static inline int deliver_skb(struct sk_buff *skb,
struct packet_type *pt_prev,
struct net_device *orig_dev)
{
if (unlikely(skb_orphan_frags_rx(skb, GFP_ATOMIC)))
return -ENOMEM;
refcount_inc(&skb->users);
return pt_prev->func(skb, skb->dev, pt_prev, orig_dev);
}
协议栈分发函数deliver_skb（）调用的pt_prev->func(skb, skb->dev,pt_prev, orig_dev)函数，是在网络协议栈初始化时赋予的指针函数，接下来再分析网络协议栈的初始化过程。