워크를 워크큐에 큐잉하는 핵심 동작은 __queue_work() 함수에서 수행합니다. 코드 분석을 통해 워크를 워크큐에 어떤 방식으로 큐잉하는지 살펴보겠습니다.
코드 분석에 앞서 __queue_work() 함수 선언부와 인자를 점검합시다.
static void __queue_work(int cpu, struct workqueue_struct *wq,
struct work_struct *work);
queue_work() 함수에서 첫 번째 인자로 WORK_CPU_UNBOUND, 두 번째 인자로 system_wq 를 전달했으니 cpu는 WORK_CPU_UNBOUND, wq는 system_wq 시스템 워크큐 전역 변수 주소입니다. 함수 인자 목록을 정리하면 다음과 같습니다.
int cpu: WORK_CPU_UNBOUND
struct workqueue_struct *wq: system_wq
struct work_struct *work: struct work_struct 구조체 주소
함수 인자를 살펴봤으니 __queue_work() 함수 처리 흐름을 단계별로 알아봅시다.
1 단계: 풀워크큐 가져오기
2 단계: 워커 구조체 가져오기
3 단계: ftrace 로그 출력하기
4 단계: 워커풀에 워크 연결 리스트 등록하고 워커 스레드 깨우기
함수 인자와 처리 단계를 알아봤으니 소스 코드를 분석할 차례입니다.
__queue_work() 함수 코드는 다음과 같습니다.
[https://github.com/raspberrypi/linux/blob/rpi-4.19.y/kernel/workqueue.c]
01 static void __queue_work(int cpu, struct workqueue_struct *wq,
02 struct work_struct *work)
03 {
04 struct pool_workqueue *pwq;
05 struct worker_pool *last_pool;
06 struct list_head *worklist;
07 unsigned int work_flags;
08 unsigned int req_cpu = cpu;
...
09 retry:
10 if (req_cpu == WORK_CPU_UNBOUND)
11 cpu = wq_select_unbound_cpu(raw_smp_processor_id());
12
13 /* pwq which will be used unless @work is executing elsewhere */
14 if (!(wq->flags & WQ_UNBOUND))
15 pwq = per_cpu_ptr(wq->cpu_pwqs, cpu);
16 else
17 pwq = unbound_pwq_by_node(wq, cpu_to_node(cpu));
18
19 last_pool = get_work_pool(work);
20 if (last_pool && last_pool != pwq->pool) {
21 struct worker *worker;
22
23 spin_lock(&last_pool->lock);
24
25 worker = find_worker_executing_work(last_pool, work);
26
27 if (worker && worker->current_pwq->wq == wq) {
28 pwq = worker->current_pwq;
30 } else {
31 /* meh... not running there, queue here */
32 spin_unlock(&last_pool->lock);
33 spin_lock(&pwq->pool->lock);
34 }
35 } else {
36 spin_lock(&pwq->pool->lock);
37 }
...
38 /* pwq determined, queue */
39 trace_workqueue_queue_work(req_cpu, pwq, work);
...
40 if (likely(pwq->nr_active < pwq->max_active)) {
41 trace_workqueue_activate_work(work);
42 pwq->nr_active++;
43 worklist = &pwq->pool->worklist;
44 if (list_empty(worklist))
45 pwq->pool->watchdog_ts = jiffies;
45 } else {
49 work_flags |= WORK_STRUCT_DELAYED;
50 worklist = &pwq->delayed_works;
51 }
52
53 insert_work(pwq, work, worklist, work_flags);
1 단계: 풀워크큐 가져오기
09번째 줄 코드부터 봅시다.
09 retry:
10 if (req_cpu == WORK_CPU_UNBOUND)
11 cpu = wq_select_unbound_cpu(raw_smp_processor_id());
req_cp 지역 변수는 이 함수에 전달되는 cpu를 그대로 저장했으니 WORK_CPU_UNBOUND입니다.
8 unsigned int req_cpu = cpu;
따라서 11번째 줄 코드를 바로 실행합니다.
11번째 줄 코드는 raw_smp_processor_id() 함수로 현재 실행 중인 CPU 번호를 알아낸 후 wq_select_unbound_cpu() 함수에 인자를 전달합니다. wq_select_unbound_cpu() 함수는 다음와 같이 동작합니다.
이진수로 1111인 wq_unbound_cpumask 비트 마스크용 변수와 AND 비트 연산으로 CPU
번호를 얻어온다.
2 단계: 워커 구조체 가져오기
이번에는 14번째 줄 코드를 보겠습니다.
14 if (!(wq->flags & WQ_UNBOUND))
15 pwq = per_cpu_ptr(wq->cpu_pwqs, cpu);
16 else
17 pwq = unbound_pwq_by_node(wq, cpu_to_node(cpu));
대부분 워크큐는 해당 플래그인 wq->flags가 0이므로 일반적인 상황에서 15번째 코드를 실행합니다. struct workqueue_struct 구조체에서 per-cpu 타입 필드인 cpu_pwqs에 접근해서 CPU 주소를 읽어 옵니다.
다음 5번째 줄 struct workqueue_struct 구조체 cpu_pwqs 필드 선언부를 보면 __percpu 키워드를 볼 수 있습니다.
[https://github.com/raspberrypi/linux/blob/rpi-4.19.y/kernel/workqueue.c]
01 struct workqueue_struct {
02 struct list_head pwqs; /* WR: all pwqs of this wq */
03 struct list_head list; /* PR: list of all workqueues */
...
04 unsigned int flags ____cacheline_aligned; /* WQ: WQ_* flags */
05 struct pool_workqueue __percpu *cpu_pwqs; /* I: per-cpu pwqs */
per-cpu 타입 변수인 wq->cpu_pwqs 필드는 다음 코드가 실행할 때 접근합니다.
15 pwq = per_cpu_ptr(wq->cpu_pwqs, cpu);
15번째 줄 코드 동작을 그림으로 표현하면 다음과 같습니다.
[그림 7.6 풀워크큐 per-cpu 자료구조]
위 그림과 같이 system_rq->cpu_pwqs 필드에 저장된 주소에 현재 실행 중인 CPU 번호 기준으로 __per_cpu_offset[cpu번호] 배열에 있는 주소를 더합니다. per-cpu 타입 변수는 CPU별로 주소 공간이 있는 것입니다. 라즈베리파이는 CPU 개수가 4개이니 4개 per-cpu 메모리 공간을 할당 받습니다.
이번에는 28줄 코드를 보겠습니다.
28 last_pool = get_work_pool(work);
struct work_struct 구조체인 work 변수로 get_work_pool() 함수를 호출합니다. 이후 struct worker_pool 구조체 주소를 last_pool 지역변수로 읽습니다. get_work_pool() 함수는 조금 후 분석할 예정입니다.
자료구조를 반경하는 코드만 읽으면 바로 이해하기 쉽지 않습니다. 다음 그림을 같이 보면서 자료구조가 어떻게 바뀌는지 알아볼까요?
[그림 7.7] 워커풀에서 워커 검색 흐름
워크를 실행한 적이 있으면 struct work_struct 구조체 data 필드는 풀워크 주소를 저장하고 있습니다. get_work_pool() 함수는 위 그림에서 [1],[2] 으로 표시된 부분과 같이 워커풀 주소를 가져옵니다.
이어서 29번째 줄 코드를 분석하겠습니다.
28 last_pool = get_work_pool(work);
29 if (last_pool && last_pool != pwq->pool) {
30 struct worker *worker;
31
32 spin_lock(&last_pool->lock);
33
34 worker = find_worker_executing_work(last_pool, work);
35
36 if (worker && worker->current_pwq->wq == wq) {
37 pwq = worker->current_pwq;
38 } else {
39 /* meh... not running there, queue here */
40 spin_unlock(&last_pool->lock);
41 spin_lock(&pwq->pool->lock);
42 }
43 } else {
44 spin_lock(&pwq->pool->lock);
45 }
29번째 줄부터 시작하는 조건문은 다음 조건을 체크합니다.
큐잉하는 워크를 이미 워커풀에서 실행한 적이 있는가?
워크를 처음 실행하면 struct work_struct 구조체 data 필드에 풀워크 주소를 저장합니다. data 필드에 액세스를 하면 워커풀 주소에 접근할 수 있습니다. 이미 워크를 처리한 풀워크가 있으면 이를 통해 워커풀을 재사용하려는 의도입니다.
다음 34번째 줄 코드를 보겠습니다.
34 worker = find_worker_executing_work(last_pool, work);
35
36 if (worker && worker->current_pwq->wq == wq) {
37 pwq = worker->current_pwq;
find_worker_executing_work() 함수를 호출해서 워크를 실행한 워커 스레드 구조체 주소를 worker 지역변수에 저장합니다. 워커 스레드 구조체 worker->current_pwq->wq에 저장된 워크큐 주소와 지금 큐잉하는 워크에 대응하는 워크큐와 같으면 struct pool_workqueue 구조체 주소를 가져옵니다. 그림 7.3에서 [3] 번호에 대응하는 동작입니다.
워크를 여러 개의 워커 스레드에서 처리하지 못하게 제약 조건을 둔 것입니다. get_work_pool() 함수와 find_worker_executing_work() 함수는 다음 소절에서 분석할 예정입니다.
3 단계: ftrace 로그 출력하기
이제 3단계 처리 과정 코드를 분석하겠습니다.
59 trace_workqueue_queue_work(req_cpu, pwq, work);
60
61 if (WARN_ON(!list_empty(&work->entry))) {
62 spin_unlock(&pwq->pool->lock);
63 return;
64 }
59번째 줄 코드는 ftrace로 workqueue_queue_work 워크큐 이벤트를 키면 실행해 다음과 같은 ftrace 로그를 출력합니다.
make-13395 [000] d.s. 589.830896: workqueue_queue_work: work struct=b9c1aa30 function=mmc_rescan workqueue=b9c06700 req_cpu=4 cpu=0
make-13395 [000] d.s. 589.830897: workqueue_activate_work: work struct b9c1aa30
위 로그는 워크를 워크큐에 큐잉했다고 해석할 수 있습니다. struct work_struct 구조체 주소는 b9c1aa30이고 워크 핸들러 함수인 mmc_rescan()입니다.
schedule_work() 함수를 호출하면 워크를 보통 워크큐에 큐잉했다고 짐작합니다. 하지만 위 ftrace 로그에서 workqueue_queue_work와 workqueue_activate_work 메시지를 확인하기 전까지 제대로 동작할 것이라 확신하면 안됩니다. 워크를 워크큐에 큐잉하는 schedule_work() 함수를 호출해도 특정 상황에서 워크가 워크큐에 제대로 큐잉을 못할 수도 있기 때문입니다.
그래서 새로운 워크를 선언한 다음 워크큐에 큐잉하는 코드를 작성했을 때 위와 같은 ftrace 메시지를 확인할 필요가 있습니다.
실력있는 개발자가 되려면 자신이 구현한 드라이버 코드가 제대로 동작하는지 점검하는 방법도 알고 있어야 합니다.
다음 68번째 줄 코드를 분석하겠습니다.
68 if (likely(pwq->nr_active < pwq->max_active)) {
69 trace_workqueue_activate_work(work);
70 pwq->nr_active++;
71 worklist = &pwq->pool->worklist;
72 if (list_empty(worklist))
73 pwq->pool->watchdog_ts = jiffies;
74 } else {
75 work_flags |= WORK_STRUCT_DELAYED;
76 worklist = &pwq->delayed_works;
77 }
먼저 if~else 문 조건을 살펴봅시다.
68번째 줄 코드는 if 조건문으로 현재 실행 중인 워커 개수를 점검합니다. if 문 결과에 따라 다음과 같이 처리합니다.
풀 워크큐에서 현재 실행 중인 워커 개수가 255를 넘지 않으면 69~73번째 줄 코드를 실행
반대의 경우 75~76번째 줄 코드를 실행
여기서 struct pool_workqueue 구조체 필드 중 nr_active는 풀 워크큐에서 현재 실행 중인 워커 개수고 max_active는 풀 워크큐에서 실행 가능한 최대 워커 개수를 의미합니다. pwq_adjust_max_active() 함수가 실행될 때 pwq->max_active는 255개로 설정됩니다.
74번째 줄 코드는 워크를 처리할 워커 개수가 255개를 넘어섰을 때 동작합니다. 이때 struct pool_workqueue 구조체 필드 중 delayed_works인 링크드 리스트를 worklist 지역 변수에 저장합니다. 여기서 보이는 delayed_works는 딜레이 워크(struct delayed_work)와 다른 개념입니다. 실행 중인 워커 개수가 많다는 것은 그만큼 처리해야 할 워크가 많다는 것을 의미합니다. 그래서 delayed_works 이란 연결 리스트(struct list_head) 필드에 워크를 등록하고 적절한 시점 이후 이 워크를 처리합니다.
딜레이 워크(struct delayed_work)는 특정 시각 후 워크를 실행합니다. 위 코드의 delayed_works와 딜레이 워크는 헷갈릴 수 있으니 주의하시길 바랍니다.
정리하면, 특정 상황에서 커널 서브 시스템이나 드라이버에서 워크를 아주 많이 생성해서 워크를 처리할 워커 개수가 255개를 넘어섰을 때를 제외하고 보통69~73번째 줄 코드를 실행합니다.
if~else 문 아래 78번째 줄 이후 코드는 69~73번 코드가 실행했다고 가정하고 분석하겠습니다.
if~else 문이 실행하는 조건을 점검했으니 69번째 줄 코드부터 분석을 시작합니다.
69번째 줄 코드는 ftrace에서 workqueue_activate_work 이벤트를 키면 다음과 같은 로그를 출력합니다. 코드와 ftrace 로그를 함께 보겠습니다.
69 trace_workqueue_activate_work(work);
make-13395 [000] d.s. 589.830896: workqueue_queue_work: work struct=b9c1aa30 function=mmc_rescan workqueue=b9c06700 req_cpu=4 cpu=0
make-13395 [000] d.s. 589.830897: workqueue_activate_work: work struct b9c1aa30
워크를 표현하는 struct work_struct 자료구조 주소를 출력하는데 보통 workqueue_queue_work ftrace 로그 이후에 출력합니다.
70번째 줄 코드는 현재 실행 중인 워커 개수를 1만큼 증감합니다.
70 pwq->nr_active++;
4 단계: 워커풀에 워크 연결 리스트 등록하고 워커 스레드 깨우기
__queue_work() 함수에서 가장 중요한 코드입니다.
71번과 76번째 줄 코드를 함께 보겠습니다.
71 worklist = &pwq->pool->worklist;
...
76 worklist = &pwq->delayed_works;
먼저 71번째 줄 코드를 분석하겠습니다.
pwq 변수는 struct pool_workqueue 구조체 타입이고 pool은 struct worker_pool 구조체 타입입니다. &pwq->pool 코드로 struct worker_pool 구조체에 접근해서 worklist 필드를 worklist 지역 변수에 저장하는 코드입니다.
이번엔 76번째 줄 코드를 보겠습니다. struct pool_workqueue 구조체인 pwq 변수로 delayed_works란 필드 주소를 worklist 지역 변수로 저장합니다.
코드만 보면 자료구조가 어떻게 바뀌는지 이해하기 어렵습니다. 다음 그림을 같이 보면서 워크를 큐잉할 때 자료구조가 어떻게 변경되는지 확인해볼까요?
[그림 7.8] 워크를 워커풀에 큐잉할 때 변경되는 자료구조
[1] &pwq->pool->worklist는 71번째 줄 [2] &pwq->delayed_works는 76번째 줄 코드를 의미합니다. 워크큐 자료 구조를 변경하는 코드를 볼 때 이렇게 전체 자료 구조 흐름을 머릿속으로 그리면서 분석하면 이해가 더 빠릅니다.
이어서 79번째 줄 코드를 보겠습니다.
79 insert_work(pwq, work, worklist, work_flags);
insert_work() 함수에 전달하는 주요 인자들을 살펴보고 insert_work() 함수 분석을 시작 하겠습니다.
pwq 는 per-cpu 타입인 struct pool_workqueue 구조체 주소를 담고 있고, work는 _queue_work() 함수로 워크큐에 큐잉하려는 워크입니다. worklist는 &pwq->pool->worklist 코드로 저장된 struct list_head 구조체인 링크드 리스트입니다.
insert_work() 함수를 호출해서 &pwq->pool->worklist 연결 리스트에 다음과 같은 워크 연결리스트를 등록합니다.
struct work_struct 구조체 entry 필드
다음 insert_work() 함수에서 wake_up_worker() 함수를 호출해서 워크를 실행할 워커 스레드를 깨웁니다.
여기까지 워크를 큐잉하는 흐름을 알아봤습니다. 다음 절에서는__queue_work_on() 함수에서 호출하는 워크큐 함수를 분석합니다.
최근 덧글