無料Wikiサービス | デモページ
検索

アクセス数
最近のコメント
kprobe - ななし
ksetの実装 - スーパーコピー
カーネルスレッドとは - ノース
カーネルスレッドとは - nbyst
asmlinkageってなに? - ノース
asmlinkageってなに? - よろしく
はじめ - ノース
はじめ - ノース
はじめ - 楽打連動ユーザー
はじめ - 楽打連動ユーザー
Adsense
広告情報が設定されていません。

ワークキュー


ワークキューはタスクレットと並んで遅延処理を行う手法です。タスクレットはそれをタスクとして起動するksoftirqdがすべてのタスクレット(他のソフト割り込みも)を担っていました。従ってどれか一つのタスクレットで遅延が発生すると、すべてに影響してしまうということです。それを回避する手段としてワークキュが誕生してということでしょうか。(たぶん)

IBMのワークキューの使い方のサンプルです。どのようになっているかと言うと、ワークキューを使用するにあたって、まずユーザが定義したworkqueue_structでcreate_workqueue関数でワークキューを作成します。そして遅延したい処理をwork_struct構造体にまとめてワークキューに登録していくと言う具合です。タスクレットではワークキューに相当するものが無く、いきなりタスクレットのリストに登録しています。ここが大きな違いで、実はワーキュキューというのはそのワークをリスト管理していると同時に、それを処理するスレッドも独自に有しているのです。しかもCPU毎に複数そのスレッドを持たせることも可能なのです。

create_workqueue(name)は__create_workqueue_key(name,0,0,&key,name)で呼ばれます。(たぶん)まずwqのworkqueue_struct構造体のkzalloc関数で確保し、wq->cpu_wqにCPU分のcpu_workqueue_struct構造体のインデックスを設定するための領域をalloc_percpu関数で割り当てています。そしてworkqueue_struct構造体でなく、このcpu_workqueue_struct構造体に実際の情報が設定されていきます。

singlethread==0の時、init_cpu_workqueue関数でのcpu_workqueue_struct構造体を初期化し、create_workqueue_thread関数でこのワークキューの処理を受け持つカーネルスレッドworker_threadを作成し、そのスレッドをcpu_workqueue_struct.threadに設定します。そしてstart_workqueue_thread関数でworker_threadを起動させています。

singlethread!=0の時、for_each_possible_cpu(cpu)で上の処理をCPU毎に行っています。CPU毎に独自のworker_threadが起動するということです。
kernel/workqueue.c
struct workqueue_struct *__create_workqueue_key(const char *name,
                                               int singlethread,
                                               int freezeable,
                                               struct lock_class_key *key,
                                               const char *lock_name)
{
       struct workqueue_struct *wq;
       struct cpu_workqueue_struct *cwq;
       int err = 0, cpu;

       wq = kzalloc(sizeof(*wq), GFP_KERNEL);
       if (!wq)
               return NULL;

       wq->cpu_wq = alloc_percpu(struct cpu_workqueue_struct);
       if (!wq->cpu_wq) {
               kfree(wq);
               return NULL;
       }

       wq->name = name;
       lockdep_init_map(&wq->lockdep_map, lock_name, key, 0);
       wq->singlethread = singlethread;
       wq->freezeable = freezeable;
       INIT_LIST_HEAD(&wq->list);

       if (singlethread) {
               cwq = init_cpu_workqueue(wq, singlethread_cpu);
               err = create_workqueue_thread(cwq, singlethread_cpu);
               start_workqueue_thread(cwq, -1);
       } else {
               cpu_maps_update_begin();
               spin_lock(&workqueue_lock);
               list_add(&wq->list, &workqueues);
               spin_unlock(&workqueue_lock);
               for_each_possible_cpu(cpu) {
                       cwq = init_cpu_workqueue(wq, cpu);
                       if (err || !cpu_online(cpu))
                               continue;
                       err = create_workqueue_thread(cwq, cpu);
                       start_workqueue_thread(cwq, cpu);
               }
               cpu_maps_update_done();
       }

       if (err) {
               destroy_workqueue(wq);
               wq = NULL;
       }
       return wq;
}
そして作成されてワークキューに遅延処理情報をまとめてwork_struct構造体を、queue_work(struct workqueue_struct *wq, struct work_struct *work)で登録することでワークが遅延処理として起動されます。 queue_work関数はそのワーキがペンディングされているかとかのチェックを行って、最終的に insert_work(cwq, work, &cwq->worklist);が呼ばれ、そこでcpu_workqueue_struct.headをルートとするリストのエンドのワークの実態をリスト追加しています。
kernel/workqueue.c
static void insert_work(struct cpu_workqueue_struct *cwq,
                       struct work_struct *work, struct list_head *head)
{
       set_wq_data(work, cwq);
       smp_wmb();
       list_add_tail(&work->entry, head);
       wake_up(&cwq->more_work);
}
ワークはworker_threadの延長線上から関数コールされるわけでが。(タスクレットと同じ。)その処理を行うのがrun_workqueue関数です。cpu_workqueue_struct.run_depth>3なら、なにやらスタックをダンプしています。あんまし遅延があって、他のCPUから起動されてりして、3回以上再エントリーしたらちょっと問題じゃない。ってユーザに警告してくれているんでしょうね?。色々とチェックとかしていますが、要はlist_empty(&cwq->worklist)で、ワークのリストが無くなるまで、work_func_t f = work->funcで設定したワーク処理を、f(work)でコールすることでワーク処理が行われるようです。
kernel/workqueue.c
static void run_workqueue(struct cpu_workqueue_struct *cwq)
{
       spin_lock_irq(&cwq->lock);
       cwq->run_depth++;
       if (cwq->run_depth > 3) {
               /* morton gets to eat his hat */
               printk("%s: recursion depth exceeded: %d\n",
                       __func__, cwq->run_depth);
               dump_stack();
       }
       while (!list_empty(&cwq->worklist)) {
               struct work_struct *work = list_entry(cwq->worklist.next,
                                               struct work_struct, entry);
               work_func_t f = work->func;
               cwq->current_work = work;
               list_del_init(cwq->worklist.next);
               spin_unlock_irq(&cwq->lock);

               BUG_ON(get_wq_data(work) != cwq);
               work_clear_pending(work);
               lock_map_acquire(&cwq->wq->lockdep_map);
               lock_map_acquire(&lockdep_map);
               f(work);
               lock_map_release(&lockdep_map);
               lock_map_release(&cwq->wq->lockdep_map);

               if (unlikely(in_atomic() || lockdep_depth(current) > 0)) {
                       printk(KERN_ERR "BUG: workqueue leaked lock or atomic: "
                                       "%s/0x%08x/%d\n",
                                       current->comm, preempt_count(),
                                       task_pid_nr(current));
                       printk(KERN_ERR "    last function: ");
                       print_symbol("%s\n", (unsigned long)f);
                       debug_show_held_locks(current);
                       dump_stack();
               }

               spin_lock_irq(&cwq->lock);
               cwq->current_work = NULL;
       }
       cwq->run_depth--;
       spin_unlock_irq(&cwq->lock);
}
singlethread==0の時(create_workqueue関数でワーク作成時)そのワークキュー内においてスリープするとその影響はそのワークキューで待つすべてのワークに影響します。従ってワークキューがスリープしてもいいと言うのは、linuxとして影響はないですが、そのあたりはユーザに責任でということでしょうか。

ワークキューはworkqueue_struct構造体を作成して、それにwork_struct構造体をリストしていました。schedule_work(struct work_struct *work)関数を使用すれば、ワークキューを作成する必要はありません。タスクレットと同じように記述することができます。理屈は簡単で、すでにカーネルとして作成しているワークキューを利用して、それを引数にqueue_work関数を呼んでいるだけです
int schedule_work(struct work_struct *work)
{
       return queue_work(keventd_wq, work);
}
こうなると考え方はタスクレットと同じです。従ってkeventd_wqのキューを使っているのは、他のユーザも使っているかもしれないので、遅延が発生するとその他のワークにも影響が発生して、他のユーザに怒られます。


追記

詳解LINUXカーネルでは、遅延処理は割り込みコンテキスト動作しますが、ワークキューの関数はプロセスコンテキストで動作する。とあります。私か理解した遅延処理とワークキューでは???です。と言うのは、遅延処理もワークキューも起動要求したところに関係なく、スケジューリングされたカーネルスレッドで起動するということだからです。それと遅延処理の起動要求は、tasklet_schedule関数を呼び出せばいいだけの話で、別に割り込みハンドラから設定する必要もないからです。また逆も言えて、ワークキューの起動要求はqueue_work関数で、これを割り込みハンドラから起動しても問題ないはずです。

確かに遅延処理は遅延処理実行後に、遅延処理が発生したかチェックして、発生したなら再度、遅延処理が実行されるようになっています。できるだけ遅延処理といえども、できるだけ早く処理をしようということなのでしょう。その点ワークキューはそのような処理はありません。(たぶん)

上の詳解LINUXカーネルの記載ですが、割り込みハンドラで使用するには、ワークキューより遅延処理の方がいいですよ。ということではないかと・・・。


最終更新 2010/09/24 03:55:38 - north
(2010/09/23 17:48:15 作成)