[리눅스 커널] Wait queue & condition

글의 참고

- https://linux-kernel-labs.github.io/refs/heads/master/so2/lec3-processes.html

- https://www.makelinux.net/ldd3/chp-6-sect-2.shtml

- https://www.cnblogs.com/hueyxu/p/13745029.html

- https://blog.csdn.net/ZHONGCAI0901/article/details/120348014

- https://blog.csdn.net/wh0604111092/article/details/78753400?spm=1001.2101.3001.6650.10&utm_medium=distribute.pc_relevant.none-task-blog-2%7Edefault%7EBlogCommendFromBaidu%7ERate-10-78753400-blog-120348014.235%5Ev38%5Epc_relevant_anti_vip&depth_1-utm_source=distribute.pc_relevant.none-task-blog-2%7Edefault%7EBlogCommendFromBaidu%7ERate-10-78753400-blog-120348014.235%5Ev38%5Epc_relevant_anti_vip&utm_relevant_index=14

- https://www.yii666.com/blog/353186.html

 

---

글의 전제

- 밑줄로 작성된 글은 강조 표시를 의미한다.

- 그림 출처는 항시 그림 아래에 표시했다.

---

글의 내용

- Wait queue

" Wait queue 는 흔히 Wait condition 이라고도 한다. 왜냐면, 특정 조건이 성립될 때 까지, 그 자리에서 잠이들기 때문이다. sleep 과 약간 비슷하지만, 분명한 차이가 존재한다. sleep 은 시간이라고 매개변수가 존재한다. 즉, `1초를 잠든다`와 같이 특정 시간동안 코드 흐름을 멈춘다. 그러나, wait queue 는 특정 조건이 참이 될 때 까지 잠이 들기 때문에,  sleep 보다 훨씬 유연하다. sleep 과 마찬가지로 wait queue 도 리눅스 커널에서 프로세스 스케줄링 메커니즘과 깊은 관련이 있다.

 

" 아래의 그림은 프로세스 상태 다이어그램을 나타낸다. 이 그림에서 우리가 다룰 부분은 wake_up() 과 wait_event() 다. 

https://linux-kernel-labs.github.io/refs/heads/master/so2/lec3-processes.html

 

 

- Wait queue data strucure

" Wait queue 자료 구조는 리눅스에서 자주 사용되는 circular linked list(struct list_head) 를 기반으로 한다. Wait queue 의 전반적인 컨트롤은 wait queue head 를 통해 이루어진다.

1. wait queue head(struct wait_queue_head) : wait queue 관리
2. wait queue entry(struct wait_queue_entry) : wait queue entry

 

" 위의 2개의 자료 구조 모두 linked list 를 기반으로 한다. wait_queue_head 를 리스트를 관리하는 자료 구조로 보고, wait_queue_entry 를 리스트에 포함되는 아이템이라고 보면 된다.

//include/linux/wait.h - v6.5
....

typedef struct wait_queue_entry wait_queue_entry_t;

typedef int (*wait_queue_func_t)(struct wait_queue_entry *wq_entry, unsigned mode, int flags, void *key);
int default_wake_function(struct wait_queue_entry *wq_entry, unsigned mode, int flags, void *key);

/* wait_queue_entry::flags */
#define WQ_FLAG_EXCLUSIVE	0x01
#define WQ_FLAG_WOKEN		0x02
#define WQ_FLAG_BOOKMARK	0x04
#define WQ_FLAG_CUSTOM		0x08
#define WQ_FLAG_DONE		0x10

/*
 * A single wait-queue entry structure:
 */
struct wait_queue_entry {
	unsigned int		flags;
	void			*private;
	wait_queue_func_t	func;
	struct list_head	entry;
};

struct wait_queue_head {
	spinlock_t		lock;
	struct list_head	head;
};
typedef struct wait_queue_head wait_queue_head_t;
....

 

" wait_queue_entry 구조체의 필드들은 다음과 같다. 

1. flags : `wait queue entry`의 상태를 나타낸다. 뒤에서 자세히 알아본다.

2. private : wait queue 는 프로세스를 wait 시키는 자료 구조다. 그리고, 나중에 조건이 트리거되면 해당 프로세스를 다시 wakeup 시켜야 한다. 이 때, `해당 프로세스`를 어떻게 깨울까? private 필드에 해당 프로세스를 저장한다. 즉, private 는 프로세스를 가리키는 포인터다.

3. func : 프로세스가 wakeup 했을 때, 호출할 callback 함수를 저장할 포인터를 의미한다.

 

" `func` 함수 포인터는 대개는 이미 커널에서 제공하는 프로세스 `wake_up` 함수를 저장한다. 이 함수 포인터를 절대 개인적으로 사용할 콜백 함수라고 생각하면 안된다. 아래에서 다루겠지만, 이 함수 포인터를 커스텀해서 사용하고 싶다면 반드시 `try_to_wake_up` 함수를 내부적으로 호출해줘야 한다.

 

" wait queue 는 리스트 구조를 사용한다고 했다. `wait_queue_head_t` 가 헤드가 되고, 그 뒤로 `wait_queue_entry_t` 들이 줄줄이 붙는 형태를 뛴다.

https://www.cnblogs.com/hueyxu/p/13745029.html

 

" 프로세스-A가 `fn`이라는 작업을 하고 싶은데, 프로세스-B가 특정 조건을 트리거 해줘야 실행할 수 있다고 치자. 그렇면, 프로세스-A는 `wait queue`에 들어가게 된다. 이제 시간이 흘러 프로세스-B 가 특정 조건을 트리거 해줬다. 그리고, `wake_up` 함수를 호출해서 `wait queue`에서 대기하고 있는 프로세스-A를 wakeup 시켰다고 치자. 그런데, 여기서 주의할 점이 있다. 프로세스-B가 wakeup 함수를 호출할 때, `wait queue`에 있는 모든 프로세스가 wakeup 된다는 것이다. 여기서 `exclusive process`라는 개념이 등장한다.

 

 

- Wait queue exculsive & thundering herd

" wait queue 안에는 항상 blocked processes 들만 존재하고 있다. 그런데, 애네가 여기에 들어온 이유는 간단한다. 다른 프로세스와 공유 자원을 가지고 경쟁하다가 점유하지 못해서 wait queue 에 들어오게 된 것이다. 그런데, wakeup 이벤트가 발생하면, wait queue 는 wakeup 이벤트를 구분해서 해당 이벤트와 대응하는 프로세스들만 개별적으로 깨우지 않는다. wait queue 에 있는 blocked processes 들을 그냥 다 깨워버린다. 즉, wakeup 에 대응하는 프로세스는 딱 하나일 수 도 있는데, wait queue 에 대기하고 있던 모든 blcoked processes 가 wakeup 된다는 것이다. 이렇게, 하나의 공유 자원을 선점하기 위해 대량의 프로세스들이 한 번에 wakeup 되는 현상을 `thundering herd` 라고 한다. 이러한 현상은 시스템 포퍼먼스에 큰 영향을 미치게 된다.

Each element in the wait queue list represents a sleeping process, which is waiting for some event to occur; its descriptor address is stored in the task field. However, it is not always convenient to wake up all sleeping processes in a wait queue.

For instance, if two or more processes are waiting for exclusive access to some resource to be released, it makes sense to wake up just one process in the wait queue. This process takes the resource, while the other processes continue to sleep. (This avoids a problem known as the "thundering herd," with which multiple processes are awoken only to race for a resource that can be accessed by one of them, and the result is that remaining processes must once more be put back to sleep.)

Thus, there are two kinds of sleeping processes: `exclusive processes` (denoted by the value 1 in the flags field of the corresponding wait queue element) are selectively woken up by the kernel, while `nonexclusive processes` (denoted by the value 0 in flags) are always woken up by the kernel when the event occurs. A process waiting for a resource that can be granted to just one process at a time is a typical exclusive process. Processes waiting for an event like the termination of a disk operation are nonexclusive.

- 참고 : https://www.halolinux.us/kernel-reference/wait-queues.html

 

" 이러한 문제를 해결하기 위해 리눅스 커널은 blocked processes 를 2개로 나눠서 관리하기 시작했다.

1. Exclusive process" 기다리고 있는 리소스가 한 시점에 딱 하나의 프로세스만 접근가능한 리소스라면, 해당 리소스에 접근하려는 프로세스들은 `exclusive process` 가 된다. 이 프로세스들은 `WQ_FLAG_EXCLUSIVE` 플래그가 설정된다. 

2. Non-exclusive process" 기다리고 있는 리소스가 여러 프로세스가 동시에 접근이 가능한 리소스일 경우, 해당 리소스에 접근하려는 프로세스들을 `non-exclusive process`라고 한다. 

 

" 정리하면, `non-exclusive processes` 들은 공유 자원에 대해 겹치는 일이없기 때문에 wakeup 시에 한 번에 다수의 프로세스들이 깨어나도 상관이 없지만, `exclusive processes` 들은 공유 자원을 같이 사용할 수 없기 때문에 wakeup 시에 딱 하나만 깨어난다[참고1]. 그렇다면, wait queue 에서는 어떻게 exclusive process 를 딱 하나만 깨울까? 뒤에서 알아본다.

 

 

- Wait queue initialization

" wait queue 초기화는 entry 와 head 초기화로 나눌 수 있다.

1. wait queue head 초기화 : DECLARE_WAIT_QUEUE_HEAD
2. wait queue entry 초기화 : DECLARE_WAITQUEUE

 

" 먼저 wait queue entry 초기화를 보자. DECLARE_WAITQUEUE() 매크로 함수를 통해서 wait queue entry 를 하나 만들 수 있다. DECLARE_WAITQUEUE() 는 생성과 초기화를 동시에 진행하는 매크로 함수다. 여기서 중요한 건 인자다. DECLARE_WAITQUEUE() 는 인자로 name, tsk 를 받는다. 즉, 이름과 wake-up 할 프로세스를 임의의로 변경이 가능하다는 뜻이다, 대신, `.func` 은 수정이 불가능하다.

// include/linux/wait.h - v6.5
....

/*
 * Macros for declaration and initialisaton of the datatypes
 */

#define __WAITQUEUE_INITIALIZER(name, tsk) {					\
	.private	= tsk,							\
	.func		= default_wake_function,				\
	.entry		= { NULL, NULL } }

#define DECLARE_WAITQUEUE(name, tsk)						\
	struct wait_queue_entry name = __WAITQUEUE_INITIALIZER(name, tsk)
    
....
....

static inline void init_waitqueue_entry(struct wait_queue_entry *wq_entry, struct task_struct *p)
{
	wq_entry->flags		= 0;
	wq_entry->private	= p;
	wq_entry->func		= default_wake_function;
}

static inline void
init_waitqueue_func_entry(struct wait_queue_entry *wq_entry, wait_queue_func_t func)
{
	wq_entry->flags		= 0;
	wq_entry->private	= NULL;
	wq_entry->func		= func;
}

....
....

 

" 그리고 init_waitqueue_*() 함수는 초기화만 해주는 함수다. 즉, 생성은 미리해야 한다. 정리하면, 다음과 같다.

1. both 생성 & 초기화 : DECLARE_*
2. only 초기화 : init_wait*()

 

" wait queue entry 를 만들 수 있는 방법은 한 가지가 더 있다. 바로, DEFINE_WAIT[_FUNC] 함수를 이용하는 것이다. DEFINE_WAIT[_FUNC] 함수는 wait queue 변수를 생성하면서, 인자로 name 과 function 을 받는다. 즉, callback 을 임의의로 수정이 가능하다. 그러나, 여기서는 대개 autoremove_wake_function() 함수를 사용한다. 이 함수는 뒤에서 다시 다룬다.

 

" init_wait() 함수는 wait queue entry 를 생성해주는 함수는 아니다. 단지, 전달받은 wait queue 를 초기화해주는 역할을 한다. 이번에 소개한 함수들은 process 를 직접 지정할 수 없다. 그렇다면, 어떤 프로세스를 blocked 시키는 걸까? 바로, `current` 프로세스를 blocked 시킨다. 즉, DEFINE_WAIT[_FUNC]() & init_wait() 함수를 통해서 wait queue entry 를 만들 경우, 해당 시점에 코드를 실행중 인 프로세스(current)를 wait queue entry 와 연결시킨다.

// kernel/sched/wait.c - v6.5
void init_wait_entry(struct wait_queue_entry *wq_entry, int flags)
{
	wq_entry->flags = flags;
	wq_entry->private = current;
	wq_entry->func = autoremove_wake_function;
	INIT_LIST_HEAD(&wq_entry->entry);
}
EXPORT_SYMBOL(init_wait_entry);
....

// include/linux/wait.h - v6.5
....

#define DEFINE_WAIT_FUNC(name, function)					\
	struct wait_queue_entry name = {					\
		.private	= current,					\
		.func		= function,					\
		.entry		= LIST_HEAD_INIT((name).entry),			\
	}

#define DEFINE_WAIT(name) DEFINE_WAIT_FUNC(name, autoremove_wake_function)

#define init_wait(wait)								\
	do {									\
		(wait)->private = current;					\
		(wait)->func = autoremove_wake_function;			\
		INIT_LIST_HEAD(&(wait)->entry);					\
		(wait)->flags = 0;						\
	} while (0)

....

 

 

" wait queue head 를 초기화하는 함수는 다음과 같다. 필드가 lock, head 밖에 없어서 상당히 간단하다. 구체적인 설명은 생략한다. wait queue head 를 생성 및 초기화를 같이하고 싶다면, DECLARE_WAIT_QUEUE_HEAD() 매크로 함수를 사용하면 되고, 초기화만 한다면, init_waitqueue_head() 함수를 호출하면 된다. 

// kernel/sched/wait.c - v6.5
void __init_waitqueue_head(struct wait_queue_head *wq_head, const char *name, struct lock_class_key *key)
{
	spin_lock_init(&wq_head->lock);
	lockdep_set_class_and_name(&wq_head->lock, key, name);
	INIT_LIST_HEAD(&wq_head->head);
}

// include/linux/wait.h - v6.5
....

#define __WAIT_QUEUE_HEAD_INITIALIZER(name) {					\
	.lock		= __SPIN_LOCK_UNLOCKED(name.lock),			\
	.head		= LIST_HEAD_INIT(name.head) }

#define DECLARE_WAIT_QUEUE_HEAD(name) \
	struct wait_queue_head name = __WAIT_QUEUE_HEAD_INITIALIZER(name)

extern void __init_waitqueue_head(struct wait_queue_head *wq_head, const char *name, struct lock_class_key *);

#define init_waitqueue_head(wq_head)						\
	do {									\
		static struct lock_class_key __key;				\
										\
		__init_waitqueue_head((wq_head), #wq_head, &__key);		\
	} while (0)
    ....

 

" wait queue 는 local 하게 관리되는 자료 구조이기 때문에, 드라이버 개발자는 DECLARE_WAIT_QUEUE_HEAD() 매크로 함수를 통해서 wait queue 를 생성하고, wait queue entry 를 `wait_event_*` 계열 함수에서 내부적으로 생성해주기 때문에, 걱정할 필요는 없다. 이 때, blocked process 는 current 프로세스가 된다.

 

" 그렇다면, 다른 프로세스를 사용하고 싶을 경우는 어떻게 할까? 커스텀 wait queue 를 만들면 된다(이 글 맨 아래 `use case` 섹션을 참고).

1. DECLARE_WAITQUEUE() 매크로 함수를 통해 별도의 프로세스를 설정한다.
2. add_wait_queue_*() 함수를 통해 wait queue entry 를 wait queue 에 저장한다.
3. set_current_state() 함수를 통해 프로세스의 상태를 block state (TASK_INTERRUPTIBLE or TASK_UNINTERRUPTIBLE) 로 변경한다.
4. wait 조건을 만든다. 조건이 false 면, schedule() 함수를 호출해서 block 하고, true 가 되면 무한 루프를 빠져나간다.

 

 

- Wait event

" wait_event_xxxxx 게열의 핵심 함수는 `___wait_event` 이다. 코드를 보면, init_wait_entry() 는 위에서 이미 언급했다시피 wait queue entry 를 초기화하는 함수다. 그리고, 무한 루프를 돌면서 condition 이 `true` 이 될 때 까지 무한루프를 돈다. condition 이 참이 되면, 무한 루프를 빠져나와 ___wait_event() 함수 이후에 코드를 이어서 실행한다.

//include/linux/wait.h - v6.5
....

/*
 * The below macro ___wait_event() has an explicit shadow of the __ret
 * variable when used from the wait_event_*() macros.
 *
 * This is so that both can use the ___wait_cond_timeout() construct
 * to wrap the condition.
 *
 * The type inconsistency of the wait_event_*() __ret variable is also
 * on purpose; we use long where we can return timeout values and int
 * otherwise.
 */

#define ___wait_event(wq_head, condition, state, exclusive, ret, cmd)		\
({										\
	__label__ __out;							\
	struct wait_queue_entry __wq_entry;					\
	long __ret = ret;	/* explicit shadow */				\
										\
	init_wait_entry(&__wq_entry, exclusive ? WQ_FLAG_EXCLUSIVE : 0);	\
	for (;;) {								\
		long __int = prepare_to_wait_event(&wq_head, &__wq_entry, state);\
										\
		if (condition)							\
			break;							\
										\
		if (___wait_is_interruptible(state) && __int) {			\
			__ret = __int;						\
			goto __out;						\
		}								\
										\
		cmd;								\
	}									\
	finish_wait(&wq_head, &__wq_entry);					\
__out:	__ret;									\
})

 

" `prepare_to_wait_event` 함수는 제일 먼저 현재 프로세스(current) 에게 pending signal 이 있는지를 확인한다[참고1 참고2]. pending signal 이 존재하면(`signal_pending_state`) 시그널을 먼저 처리해야 하기 때문에, `wait queue`에서 프로세스를 깨우지 않고 함수를 종료한다.

 

" 만약, 현재 프로세스에 pending signal 이 없다면, 제일 머저 wait queue entry 가 이미 wait queue 에 삽입되어 있는지 검사한다. list_empty() 함수를 통해 wait queue entry 가 추가 여부를 검사하는 이유는 wait queue entry 를 딱 한 번만 추가하기 위해서다. ___wait event() 함수는 무한 루프를 통해서 prepare_to_wait_event() 함수를 계속 호출한다. 그런데, 그 때마다 wait queue entry 를 추가하면 안된다. 그래서, wait queue entry 가 리스트에 추가되면 entry->next 는 빈 값이 아니게 되기 때문에, list_empty() 에서 false 를 반환한다. 만약, wait queue 에 연결되어 있지않다면, wait queue 에 추가하는 루틴을 타게된다.

// kernel/sched/wait.c - v6.5
long prepare_to_wait_event(struct wait_queue_head *wq_head, struct wait_queue_entry *wq_entry, int state)
{
	unsigned long flags;
	long ret = 0;

	spin_lock_irqsave(&wq_head->lock, flags);
	if (signal_pending_state(state, current)) {
		/*
		 * Exclusive waiter must not fail if it was selected by wakeup,
		 * it should "consume" the condition we were waiting for.
		 *
		 * The caller will recheck the condition and return success if
		 * we were already woken up, we can not miss the event because
		 * wakeup locks/unlocks the same wq_head->lock.
		 *
		 * But we need to ensure that set-condition + wakeup after that
		 * can't see us, it should wake up another exclusive waiter if
		 * we fail.
		 */
		list_del_init(&wq_entry->entry);
		ret = -ERESTARTSYS;
	} else {
		if (list_empty(&wq_entry->entry)) {
			if (wq_entry->flags & WQ_FLAG_EXCLUSIVE)
				__add_wait_queue_entry_tail(wq_head, wq_entry);
			else
				__add_wait_queue(wq_head, wq_entry);
		}
		set_current_state(state);
	}
	spin_unlock_irqrestore(&wq_head->lock, flags);

	return ret;
}
EXPORT_SYMBOL(prepare_to_wait_event);

 

" 이제 부터 중요한 내용이다. wait queue 에 들어갈 때, wait queue entry 가 어떤 `속성` 을 가지고 있냐에 따라 wait queue 에 삽입되는 위치가 달라진다. 그 기준은 wait queue entry 에 `WQ_FLAG_EXCLUSIVE` 플래그가 설정되어있는지 여부에 따라 달라진다.

1. 만약, WQ_FLAG_EXCLUSIVE 플래그가 설정되어 있으면, wait queue에 뒷쪽에 추가한다.
2. 만약, WQ_FLAG_EXCLUSIVE 플래그가 설정되어 있지 않으면, wait queue에 앞쪽에 추가한다.

 

" WQ_FLAG_EXCLUSIVE 플래그 여부에 따라 삽입되는 위치가 달라지는 이유가 뭘까? 나중에 보겠지만, `wake_up` 함수는 `wait queue` 를 앞에서부터 순회하면서 모든 `non-exclusive processes` 들을 깨우고, 첫 번째로 발견되는 `exclusive process`를 하나만 깨우고 루프를 종료한다. 이해가 안간다면 `Wait queue exculsive & thundering herd` 섹션을 다시 읽어보자.

 

" 그리고, 마지막으로 프로세스의 상태를 `TASK_INTERRUTIBLE` 혹은 `TASK_UNINTERRUPTIBLE` 로 변경한다(`set_current_state`).

 

" `__add_wait_queue_xxxx` 계열에서 가장 핵심적인 함수는 `__add_wait_queue` 함수다. 이 함수를 통해 저장되는 프로세스들은 `wait queue` 안에서 아래의 구조를 띄고 있다.

1. 제일 앞쪽에 WQ_FLAG_PRIORITY 가 SET 된 프로세스들이 저장된다(`__wake_up_process` 함수가 정의되는 주석에 설명되어 있다).
2. 제일 뒤쪽에 WQ_FLAG_EXCLUSIVE 가 SET 된 프로세스들이 저장된다.
3. WQ_FLAG_PRIORITY 와 WQ_FLAG_EXCLUSIVE 사이에 `non-exclusive processes` 들이 저장된다.

 

" `list_add` 함수는 첫 번째 인자를 두 번째 인자 바로 뒤에 추가한다. `list_add_tail` 함수는 첫 번째 인자를 두 번째 인자 바로 앞에 추가한다.

// include/linux/wait.h - V6.5
....

static inline void __add_wait_queue(struct wait_queue_head *wq_head, struct wait_queue_entry *wq_entry)
{
	struct list_head *head = &wq_head->head;
	struct wait_queue_entry *wq;

	list_for_each_entry(wq, &wq_head->head, entry) {
		if (!(wq->flags & WQ_FLAG_PRIORITY))
			break;
		head = &wq->entry;
	}
	list_add(&wq_entry->entry, head);
}

/*
 * Used for wake-one threads:
 */
static inline void
__add_wait_queue_exclusive(struct wait_queue_head *wq_head, struct wait_queue_entry *wq_entry)
{
	wq_entry->flags |= WQ_FLAG_EXCLUSIVE;
	__add_wait_queue(wq_head, wq_entry);
}

static inline void __add_wait_queue_entry_tail(struct wait_queue_head *wq_head, struct wait_queue_entry *wq_entry)
{
	list_add_tail(&wq_entry->entry, &wq_head->head);
}

// kernel/sched/wait.c - v6.5
void add_wait_queue(struct wait_queue_head *wq_head, struct wait_queue_entry *wq_entry)
{
	unsigned long flags;

	wq_entry->flags &= ~WQ_FLAG_EXCLUSIVE;
	spin_lock_irqsave(&wq_head->lock, flags);
	__add_wait_queue(wq_head, wq_entry);
	spin_unlock_irqrestore(&wq_head->lock, flags);
}
EXPORT_SYMBOL(add_wait_queue);

void add_wait_queue_exclusive(struct wait_queue_head *wq_head, struct wait_queue_entry *wq_entry)
{
	unsigned long flags;

	wq_entry->flags |= WQ_FLAG_EXCLUSIVE;
	spin_lock_irqsave(&wq_head->lock, flags);
	__add_wait_queue_entry_tail(wq_head, wq_entry);
	spin_unlock_irqrestore(&wq_head->lock, flags);
}
EXPORT_SYMBOL(add_wait_queue_exclusive);

void add_wait_queue_priority(struct wait_queue_head *wq_head, struct wait_queue_entry *wq_entry)
{
	unsigned long flags;

	wq_entry->flags |= WQ_FLAG_EXCLUSIVE | WQ_FLAG_PRIORITY;
	spin_lock_irqsave(&wq_head->lock, flags);
	__add_wait_queue(wq_head, wq_entry);
	spin_unlock_irqrestore(&wq_head->lock, flags);
}
EXPORT_SYMBOL_GPL(add_wait_queue_priority);

 

" `add_wait_queue_priority` 함수는 우선순위가 높은 엔트리를 저장하는 함수다. 우선순위가 높은 엔트리는 wait queue 의 제일 앞쪽부터 저장된다(`WQ_FLAG_PRIORITY`). 그리고, `WQ_FLAG_EXCLUSIVE` 플래그도 같이 SET 되어있기 때문에, wake_up() 함수 호출시, `WQ_FLAG_PRIORITY` 플래그 SET 된 노드의 프로세스만 깨우고 wake_up() 함수를 종료한다.

 

" `finish_wait` 함수는 현재 동작하고 있는 프로세스를 `TASK_RUNNING` 상태로 바꾼 뒤, 엔트리를 wait queue 에서 제거한다. finish_wait() 함수가 호출되었다는 것은 condition 이 참이 되었다는 것을 의미한다. 즉, wait queue 에 엔트리를 넣을 필요가 없다는 뜻이다. 그런데, `prepare_to_wait_event`함수에서 이미 엔트리를 `wait queue`에 넣고 프로세스의 상태마저 변경해버렸다. 그러므로, 앞에서 행한 일들을 원상태로 복구시켜야 한다.

/**
 * finish_wait - clean up after waiting in a queue
 * @wq_head: waitqueue waited on
 * @wq_entry: wait descriptor
 *
 * Sets current thread back to running state and removes
 * the wait descriptor from the given waitqueue if still
 * queued.
 */
void finish_wait(struct wait_queue_head *wq_head, struct wait_queue_entry *wq_entry)
{
	unsigned long flags;

	__set_current_state(TASK_RUNNING);
	/*
	 * We can check for list emptiness outside the lock
	 * IFF:
	 *  - we use the "careful" check that verifies both
	 *    the next and prev pointers, so that there cannot
	 *    be any half-pending updates in progress on other
	 *    CPU's that we haven't seen yet (and that might
	 *    still change the stack area.
	 * and
	 *  - all other users take the lock (ie we can only
	 *    have _one_ other CPU that looks at or modifies
	 *    the list).
	 */
	if (!list_empty_careful(&wq_entry->entry)) {
		spin_lock_irqsave(&wq_head->lock, flags);
		list_del_init(&wq_entry->entry);
		spin_unlock_irqrestore(&wq_head->lock, flags);
	}
}
EXPORT_SYMBOL(finish_wait);

 

 

" 가장 많이 사용하는 wait_event() 는 내부적으로 __wait_event() 함수를 호출하고, wait_event_interruptible() 는 내부적으로 __wait_event_interruptible() 함수를 호출한다. 중요하게 볼 부분은 마지막 인자로 `schedule()` 함수를 넣었다는 것이다.

// include/linux/wait.h - v6.5
#define __wait_event(wq_head, condition)					\
	(void)___wait_event(wq_head, condition, TASK_UNINTERRUPTIBLE, 0, 0,	\
			    schedule())

....
#define __wait_event_interruptible(wq_head, condition)				\
	___wait_event(wq_head, condition, TASK_INTERRUPTIBLE, 0, 0,		\
		      schedule())

 

 

 

- Wake up

" `wait_event`에 짝을 이루는 함수가 `wake_up` 함수다. `wait queue`에서 대기하고 있던 프로세스를 깨워 멈춰있던 코드를 다시수행하도록 하는 함수다.

 

" `wak_up_xxx` 계열 함수에서 원형은 `__wake_up_common` 함수다. 이 함수는 처음부터 루프를 탐색하지 않고, curr(최초의 엔트리일 수 도 있고, bookmark 가 SET 되어 있으면 아닐 수 도 있다) 노드부터 탐색한다. next 는 curr 의 다음 entry 를 의미한다. 뒤에서 보겠지만, bookmark entry 는 다음에 어디부터 실행할지를 판단하는데 기준이 되는 entry 다. 대신 아무 기능이 없는 entry 다. 즉, bookmark entry 는 다음 탐색시에 어디서부터 실행하면 되는 지표가 되어주지만, 기능을 가지고 있지는 않다. 그렇기 때문에, bookmark entry 가 나타나면 `continue` 명령어를 통해 뒤에 명령어를 실행하지 않고, 다음 노드로 jump 한다.

 

" 그리고, 마침내 curr->func 코드가 실행되면, wait queue entry 에 저장되어 있던 프로세스(`curr->private`) 를 wait queue 에서 꺼낸 뒤, 스케줄링 대상으로 변경시킨다(`TASK_RUNNING`). curr->func 작업을 마치고 나면, 해당 프로세스가 exclusive process 이고, 더 이상 exclusive process 가 존재하지 않는다면, 루프를 종료한다. 즉, wait queue 탐색은 첫 번째 exclusive process 를 wakeup 시키는 순간 종료된다. 만약, non-exclusive process 라면 계속 루프를 탐색해서 blocked processes 들을 wakeup 시키게 된다. 만약에 wait queue 에 non-exclusive process 만 존재한다면, `wait queue`를 처음부터 끝까지 탐색하게 된다. 즉, list_for_each_entry_safe_from 함수 내부에서 탈출 조건으로 종료되는 것이 아니라, 이 함수 자체가 종료된다.

// include/linux/wait.h - v6.5
....

#define wake_up(x)			__wake_up(x, TASK_NORMAL, 1, NULL)
#define wake_up_nr(x, nr)		__wake_up(x, TASK_NORMAL, nr, NULL)
#define wake_up_all(x)			__wake_up(x, TASK_NORMAL, 0, NULL)
....

#define wake_up_interruptible(x)	__wake_up(x, TASK_INTERRUPTIBLE, 1, NULL)
#define wake_up_interruptible_nr(x, nr)	__wake_up(x, TASK_INTERRUPTIBLE, nr, NULL)
#define wake_up_interruptible_all(x)	__wake_up(x, TASK_INTERRUPTIBLE, 0, NULL)

// kernel/sched/wait.c - v6.5
/*
 * The core wakeup function. Non-exclusive wakeups (nr_exclusive == 0) just
 * wake everything up. If it's an exclusive wakeup (nr_exclusive == small +ve
 * number) then we wake that number of exclusive tasks, and potentially all
 * the non-exclusive tasks. Normally, exclusive tasks will be at the end of
 * the list and any non-exclusive tasks will be woken first. A priority task
 * may be at the head of the list, and can consume the event without any other
 * tasks being woken.
 *
 * There are circumstances in which we can try to wake a task which has already
 * started to run but is not in state TASK_RUNNING. try_to_wake_up() returns
 * zero in this (rare) case, and we handle it by continuing to scan the queue.
 */
static int __wake_up_common(struct wait_queue_head *wq_head, unsigned int mode,
			int nr_exclusive, int wake_flags, void *key,
			wait_queue_entry_t *bookmark)
{
	wait_queue_entry_t *curr, *next;
	int cnt = 0;

	lockdep_assert_held(&wq_head->lock);

	if (bookmark && (bookmark->flags & WQ_FLAG_BOOKMARK)) { // --- 1
		curr = list_next_entry(bookmark, entry);

		list_del(&bookmark->entry);
		bookmark->flags = 0;
	} else
		curr = list_first_entry(&wq_head->head, wait_queue_entry_t, entry); // --- 1

	if (&curr->entry == &wq_head->head) // --- 2
		return nr_exclusive;

	list_for_each_entry_safe_from(curr, next, &wq_head->head, entry) {
		unsigned flags = curr->flags;
		int ret;

		if (flags & WQ_FLAG_BOOKMARK) // --- 3
			continue;

		ret = curr->func(curr, mode, wake_flags, key); // --- 4
		if (ret < 0)
			break;
		if (ret && (flags & WQ_FLAG_EXCLUSIVE) && !--nr_exclusive) // --- 5
			break;

		if (bookmark && (++cnt > WAITQUEUE_WALK_BREAK_CNT) && // --- 6
				(&next->entry != &wq_head->head)) {
			bookmark->flags = WQ_FLAG_BOOKMARK;
			list_add_tail(&bookmark->entry, &next->entry);
			break;
		}
	}

	return nr_exclusive;
}


/**
 * __wake_up - wake up threads blocked on a waitqueue.
 * @wq_head: the waitqueue
 * @mode: which threads
 * @nr_exclusive: how many wake-one or wake-many threads to wake up
 * @key: is directly passed to the wakeup function
 *
 * If this function wakes up a task, it executes a full memory barrier
 * before accessing the task state.  Returns the number of exclusive
 * tasks that were awaken.
 */
int __wake_up(struct wait_queue_head *wq_head, unsigned int mode,
	      int nr_exclusive, void *key)
{
	return __wake_up_common_lock(wq_head, mode, nr_exclusive, 0, key);
}
EXPORT_SYMBOL(__wake_up);

 

1. 초기 조건문의 내용은 wait queue 에 bookmark 가 표시되어 있는지 여부에 따라 `curr` 값을 바꾸는 것이다.

1. bookmark 존재 - 북마크가 존재하면, 북마크 노드 다음 노드가 탐색을 시작할 노드가 된다. 2. bookmark 미존재 - 북마크가 존재하지 않으면, 헤드의 가장 첫 번째 노드를 가져오게 된다. 즉, 처음부터 탐색

" 여기서 bookmark 란 wait queue 가 CPU 를 독점하지 못하게 막는 것이라 보면 된다. 만약, wait queue 에 너무 많은 프로세스가 대기하고 있으면, 모든 프로세스를 깨울 때 까지 루프를 반복하게 된다. 이러한 현상을 막기 위해 break count 를 설정하고, 어디까지 깨웠는지를 저장한다. 이 때, `어디까지 깨웠는지를 저장` 에 해당하는 내용이 bookmark 다.

2. 그리고, curr 이 wait queue head 와 동일하다면 wait queue 가 비어있다는 뜻이므로, wakeup 루틴을 종료한다.

3. bookmark entry 는 dummy entry 다. 단지, 위치 표시로 사용되므로, skip 한다.

4. 일반적으로, curr->func 에 등록되는 callback 함수는 default_wake_function 혹은 autoremove_wake_function 함수다. 이 함수들은 최종적으로 `try_to_wake_up` 함수를 호출하는데, 여기서 프로세스는 TASK_RUNNING 상태로 변경된다. 

5. nr_exclusive 인자는 exclusive process 를 몇 개 깨울 건지를 나타낸다. 분명히 위에서는 exclusive process 하나만 깨우고 루프를 종료한다고 했지만, 커널이 점점 발전하는 것 같다. 사용자들에 니즈를 충족시키기 위해 wake-up 시킬 exclusive process 개수까지 설정할 수 있게 됬다. 그래서, `wake_up_all` 매크로 함수를 보면, nr_exclusive 로 `0` 을 전달하는데, 이렇게 되면 wait queue 에 있는 모든 프로세스들을 깨우게 된다. 왜냐면, nr_exclusive 이 0 이면, 전위 감소 연산자와 만나 -1 이 되고, -1 이 `!` 연산자를 만나면, false 가 된다. 그 이후에 전위 감소 연산자 때문에 계속 음수가 되고, 이는 계속해서 false 를 반환하게 되는 상황이 된다.

그런데, non-exclusive process 는 exclusive process 를 만날 때 까지, 계속 wake-up 한다고 했는데 도대체 어디에 그 코드가 있을까? 바로 `flags & WQ_EXCLUSIVE` 다. non-exclusive process 들은 이 조건문에서 참이되지 못하게 때문에, 계속해서 루프를 돌게된다.

6. 앞에서 wait queue 가 CPU 를 독점하는 것을 막기위해 break count 와 bookmark 를 이용한다고 했다. break count 의 조건은 아래와 같다. 3개 모두 참이어야 한다[참고1 참고2].

1. bookmark 리스트가 정상적으로 초기화가 되었는지 2. 한 번에 깨어날 수 있는 wakeup processes 개수(`WAITQUEUE_WALK_BREAK_CNT` == 64)를 초과했는지 3. wait queue 가 비어있지 않은지

" 위 3개의 조건이 충족되면, 다음으로 wakeup 할 노드(`next->entry`) 앞에 북마크 노드(`bookmark->entry`)를 추가한다. 이제 `__wake_up_common` 함수의 처음 조건문을 다시보자(list_add_tail() 함수의 두 번째 인자가 `head` 고, 첫 번째 인자가 `new` 다).

 

 

" 그리고, bookmark 관련 또 눈여겨 볼 점이 __wake_up_common 함수를 래핑하는 __wake_up_common_lock() 함수 내부를 보면, bookmark entry 에 WQ_FLAG_BOOKMARK 플래그 SET 되어있으면, 다시 _wake_up_common() 함수를 호출한다. 즉, `wait queue 에 아직 blocked process 들이 남아있는데, break count 때문에 잠깐 휴식취하고 다시 wakeup 시키러 가야한다` 의 의미로 보면된다. 그런데, wait queue 에 exclusive process 가 존재하지 않고, non-exclusive process 만 존재한다면, wait queue empty 가 될 때 까지 프로세스들을 wakeup 시킬 수 있다. 이건 문제가 있어보인다.

// kernel/sched/wait.c - v6.5
static int __wake_up_common_lock(struct wait_queue_head *wq_head, unsigned int mode,
			int nr_exclusive, int wake_flags, void *key)
{
	unsigned long flags;
	wait_queue_entry_t bookmark;
	int remaining = nr_exclusive;

	bookmark.flags = 0;
	bookmark.private = NULL;
	bookmark.func = NULL;
	INIT_LIST_HEAD(&bookmark.entry);

	do {
		spin_lock_irqsave(&wq_head->lock, flags);
		remaining = __wake_up_common(wq_head, mode, remaining,
						wake_flags, key, &bookmark);
		spin_unlock_irqrestore(&wq_head->lock, flags);
	} while (bookmark.flags & WQ_FLAG_BOOKMARK);

	return nr_exclusive - remaining;
}

 

" 만약, WAITQUEUE_WALK_BREAK_CNT 값에 끊지 않는다면, 꾀나 문제가 되는 상황이 올 수 있다. 왜냐면, __wake_up_common 함수가 `spin_lock_irqsave` / `spin_lock_irqrestore` 함수들로 감싸져있기 때문이다. 이 함수들은 프로세스 선점도 막고, 인터럽트도 비활성화한다. 즉, __wake_up_common() 함수는 SMP 환경에서도 안전하게 실행된다는 것이다. 그런데, wait queue 에 100000개 프로세스가 들어있다면 어떻게 될까? 저 많은 프로세스를 실행시킬 때 까지 wait queue 를 한 프로세스가 선점하게 된다. 이건 퍼포먼스에 영향을 주게된다. 그래서, 탐색은 WAITQUEUE_WALK_BREAK_CNT 값보다 커지면 끊고, 다음 루프때를 기다린다(그렇다고, `schedule` 함수를 호출해서 CPU를 양보하는 것은 아니다).

 

" 그런데, 어차피 try_to_wake_up() 함수는 TASK_RUNNING 상태만 변경시키기 때문에, 크게 오버헤드가 있을 것 같지는 않다. 그런데도, 조금 걱정이 되는 부분은 spinklock 이 global lock 이라는 점이다.

 

" `default_wake_function` 함수와 `autoremove_wake_function` 함수는 결국 내부적으로 `try_to_wake_up` 함수를 호출한다. 즉, `try_to_wake_up` 함수는 sleeping 혹은 blocked 된 프로세스를 wakeup 시켜서 `TASK_RUNNING` 상태로 변경한 뒤, CPU 선택 알고리즘을 통해 특정 CPU 의 런큐에 할당된다. 스케줄러에 의해 해당 프로세스가 스케줄링되면 이전에 멈춰있었던 wait condition 상황부터 코드를 다시 시작할 수 있게된다. 즉, wake-up 하고 다시 `condition` 값을 검사하고 아직도 false 면 schedule() 함수를 통해서 다시 blocked 상태가 된다.

// kernel/sched/core.c - v6.5
....

int default_wake_function(wait_queue_entry_t *curr, unsigned mode, int wake_flags,
			  void *key)
{
	WARN_ON_ONCE(IS_ENABLED(CONFIG_SCHED_DEBUG) && wake_flags & ~WF_SYNC);
	return try_to_wake_up(curr->private, mode, wake_flags);
}
EXPORT_SYMBOL(default_wake_function);
....

// kernel/sched/wait.c - v6.5
int autoremove_wake_function(struct wait_queue_entry *wq_entry, unsigned mode, int sync, void *key)
{
	int ret = default_wake_function(wq_entry, mode, sync, key);

	if (ret)
		list_del_init_careful(&wq_entry->entry);

	return ret;
}
EXPORT_SYMBOL(autoremove_wake_function);

 

" `try_to_wake_up` 함수는 SMP 에 대한 예외 처리 때문에 코드가 상당히 복잡하다. 그중에서 CPU 마이그레이션 관련 코드가 있는데, 관련된 내용은 프로세스쪽에서 자세히 다루도록 한다. 그리고, 주의할 점이 있다. default_wake_function() 함수는 wait queue entry() 를 리스트에서 지우지 않는다. 상식적으로는 사용했다면 해제하는것이 맞다. 그러므로, autoremove_wake_function 함수를 사용해서 자동으로 제거되도록 한다. 

 

 

- Use case

" 테스크 코드를 작성해서 실제로 어떻게 `wait_event` & `wake_up` 을 사용하는지 알아보자. 먼저, 유저 애플리케이션에서 GPIO의 값을 읽어갈 수 있도록 `gpio_key_read` 함수를 작성한다. 이 함수는 `/dev/gpio_key0` 파일을 읽을 때, 호출된다.

// https://blog.csdn.net/ZHONGCAI0901/article/details/120348014
....

static DECLARE_WAIT_QUEUE_HEAD(gpio_key_wait);
bool key_event_flag;
....

ssize_t gpio_key_read(struct file *pFile, char __user *pBuf, size_t size, loff_t *ppos)
{
    ssize_t len;
    int err;
    
    wait_event_interruptible(gpio_key_wait, key_event_flag);
    
    len = sizeof(struct app_key_event);
    err = copy_to_user(pBuf, &appKeyPayload, len);
    
    key_event_flag = false;
    
    return len;
}

 

" `gpio_key_read` 함수 내부를 보면, 내부적으로 `wait_event_interruptible` 함수가 호출되는 것을 볼 수 있다. `gpio_key_read` 함수는 `wait_event_interruptible` 함수를 만나는 순간, `key_event_flag`가 TRUE가 될 때 까지 BLOCK 상태가 된다. 즉, 두 번째 인자가 TRUE가 되어야, 그 다음 코드인 `len = sizeof ...` 를 실행할 수 있게 된다.

 

" 이제 GPIO 버튼을 누르면, 인터럽트를 발생시키고 인터럽트 핸들러를 동작시키도록 하자. 그리고, 핸들러안에서 `key_event_flag` 를 TRUE 로 변경하자. 그렇면, BLOCK 된 프로세스가 트리거되면서 스케줄링 대상으로 변경될 것 이다.

// https://blog.csdn.net/ZHONGCAI0901/article/details/120348014
static irqreturn_t gpio_key_isr(int irq, void *dev_id)
{
	int val;
    struct gpio_key_cfg *pGpioKey = dev_id;
    
    val = gpiod_get_value(pGpioKey->gpiod);
    
    printk("key %d %d\n", pGpioKey->gpio_num, val);
    
    appKeyPayload.gpio_num = pGpioKey->gpio_num;
    appKeyPayload.value = val;
    
    key_event_flag = true;
    
    wake_up_interruptible(&gpio_key_wait);
    
    return IRQ_HANDLED;
}

 

" 그런데, 이상한 점이 있다. gpio_key_read() 함수에서 key_event_flag 의 주소가 아닌 `값` 을 전달한다. wait queue 는 값만 전달받는데 어떻게 key_event_flag 의 값이 변경했다는 것을 알 수 있을까? wait queue 관련 대부분의 함수들이 매크로로 작성 되어있다. 즉, 컴파일 시점에 전처리를 통해서 wait queue 매크로 함수들이 코드로 전환된다. 예를 들어, gpio_key_read() 함수는 전처리 과정을 거치면 아래와 같이 변환된다.

 

// https://blog.csdn.net/ZHONGCAI0901/article/details/120348014
....

static DECLARE_WAIT_QUEUE_HEAD(gpio_key_wait);
bool key_event_flag;
....

ssize_t gpio_key_read(struct file *pFile, char __user *pBuf, size_t size, loff_t *ppos)
{
    ssize_t len;
    int err;
    
    key_event_flag = false;
    ({										\
	__label__ __out;							\
	struct wait_queue_entry __wq_entry;					\
	long __ret = 0;	/* explicit shadow */				\
										\
	init_wait_entry(&__wq_entry, 0 ? WQ_FLAG_EXCLUSIVE : 0);	\
	for (;;) {								\
		long __int = prepare_to_wait_event(&gpio_key_wait, &__wq_entry, TASK_INTERRUPTIBLE);\
										\
		if (key_event_flag)							\
			break;							\
										\
		if (___wait_is_interruptible(TASK_INTERRUPTIBLE) && __int) {			\
			__ret = __int;						\
			goto __out;						\
		}								\
										\
		schedule();								\
	}									\
	finish_wait(&gpio_key_wait, &__wq_entry);					\
	__out:	__ret;									\
    })
    
    len = sizeof(struct app_key_event);
    err = copy_to_user(pBuf, &appKeyPayload, len);
    
    return len;
}

 

" 매크로가 코드로 치환되면서, 왜 주소를 전달하지 않았는지 알 수 있게 됬다. 그리고, 위에서 wait_up_interruptible() 함수는 사실 호출할 필요는 없다. 단지 조금 더 빨리 깨울뿐이다.

 

" 또 다른 예제를 하나 보자. `MMC` 드라이버 코드를 들고 와봤다. MMC에서는 `wait_event_xxx` 함수를 사용하지 않는다. `wait_event` 함수를 기능이 엔트리를 생생해서 `wait queue`에 넣은 뒤, condition이 참이 될 때 까지 무한 루프를 도는 것이다. 그런ㄷ, MMC는 이걸 자기가 직접 구현하고 있다. 아래 코드에서, `__mmc_claim_host` 함수안에 3가지 내용이 보인다.

1. wait queue entry 생성
2. 무한 루프 & 탈출 조건
3. schedule 함수

 

" `MMC`는 왜 `wait_event_xxx` 함수를 사용하지 않았을까? 일반적으로, `wait_event_xxx` 함수는 `condition` 을 하나만 걸 수 있다. 그런데, MMC는 컨디션이 무려 3개나 걸려있다(`if (stop || !host->claimed || mmc_ctx_matches(host, ctx, task))!). 즉, 무한 루프를 탈출하기 위한 컨디션을 1개 이상 만들고 싶다면, 아래와 같이 커스텀한 방식으로 `wait queue` 를 구현할 수 있다. 

int __mmc_claim_host(struct mmc_host *host, struct mmc_ctx *ctx,
		     atomic_t *abort)
{
	struct task_struct *task = ctx ? NULL : current;
	DECLARE_WAITQUEUE(wait, current);
	unsigned long flags;
	int stop;
	bool pm = false;

	might_sleep();

	add_wait_queue(&host->wq, &wait);
	spin_lock_irqsave(&host->lock, flags);
	while (1) {
		set_current_state(TASK_UNINTERRUPTIBLE);
		stop = abort ? atomic_read(abort) : 0;
		if (stop || !host->claimed || mmc_ctx_matches(host, ctx, task))
			break;
		spin_unlock_irqrestore(&host->lock, flags);
		schedule();
		spin_lock_irqsave(&host->lock, flags);
	}
	set_current_state(TASK_RUNNING);
	....
    
	spin_unlock_irqrestore(&host->lock, flags);
	....
    
	return stop;
}

void mmc_release_host(struct mmc_host *host)
{
	....
	if (--host->claim_cnt) {
		....
	} else {
		host->claimed = 0;
		host->claimer->task = NULL;
		host->claimer = NULL;
		spin_unlock_irqrestore(&host->lock, flags);
		wake_up(&host->wq);
		....
	}
}

 

" 아마, 그냥 `chk = (stop || !host->claimed || mmc_ctx_matches(host, ctx, task)); if(chk) break;` 로 조건을 하나로 merge 해서 wait_event() 함수를 사용되지 않을까?` 라는 생각을 할 수 도 있다. 그러나, 이렇게 하면 모든 조건을 다 체크하게 된다. 예를 들어, 일일히 조건을 검사하면, stop 이 `1` 만 되더라도 if 문이 true 가 된다. 그러나, 3개의 조건을 하나에 변수에 merge 하면 3개의 결과를 다 뽑아야 해서 성능면에서 좋지 못할 수 있다.

 

" `mmc_release_host` 함수에서는 `claim`이 없을 경우, 무한 루프 탈출 조건을 생성한 뒤(`host->claimed = 0`), `wake_up` 함수를 호출해서 `__mmc_claim_host` 함수에서 선점되었던 프로세스를 다시 런큐에 삽입시킨다.