2 * padata.c - generic interface to process data streams in parallel
4 * Copyright (C) 2008, 2009 secunet Security Networks AG
5 * Copyright (C) 2008, 2009 Steffen Klassert <steffen.klassert@secunet.com>
7 * This program is free software; you can redistribute it and/or modify it
8 * under the terms and conditions of the GNU General Public License,
9 * version 2, as published by the Free Software Foundation.
11 * This program is distributed in the hope it will be useful, but WITHOUT
12 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
13 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
16 * You should have received a copy of the GNU General Public License along with
17 * this program; if not, write to the Free Software Foundation, Inc.,
18 * 51 Franklin St - Fifth Floor, Boston, MA 02110-1301 USA.
21 #include <linux/module.h>
22 #include <linux/cpumask.h>
23 #include <linux/err.h>
24 #include <linux/cpu.h>
25 #include <linux/padata.h>
26 #include <linux/mutex.h>
27 #include <linux/sched.h>
28 #include <linux/rcupdate.h>
30 #define MAX_SEQ_NR INT_MAX - NR_CPUS
31 #define MAX_OBJ_NUM 10000 * NR_CPUS
33 static int padata_index_to_cpu(struct parallel_data
*pd
, int cpu_index
)
37 target_cpu
= cpumask_first(pd
->cpumask
);
38 for (cpu
= 0; cpu
< cpu_index
; cpu
++)
39 target_cpu
= cpumask_next(target_cpu
, pd
->cpumask
);
44 static int padata_cpu_hash(struct padata_priv
*padata
)
47 struct parallel_data
*pd
;
52 * Hash the sequence numbers to the cpus by taking
53 * seq_nr mod. number of cpus in use.
55 cpu_index
= padata
->seq_nr
% cpumask_weight(pd
->cpumask
);
57 return padata_index_to_cpu(pd
, cpu_index
);
60 static void padata_parallel_worker(struct work_struct
*work
)
62 struct padata_queue
*queue
;
63 struct parallel_data
*pd
;
64 struct padata_instance
*pinst
;
65 LIST_HEAD(local_list
);
68 queue
= container_of(work
, struct padata_queue
, pwork
);
72 spin_lock(&queue
->parallel
.lock
);
73 list_replace_init(&queue
->parallel
.list
, &local_list
);
74 spin_unlock(&queue
->parallel
.lock
);
76 while (!list_empty(&local_list
)) {
77 struct padata_priv
*padata
;
79 padata
= list_entry(local_list
.next
,
80 struct padata_priv
, list
);
82 list_del_init(&padata
->list
);
84 padata
->parallel(padata
);
91 * padata_do_parallel - padata parallelization function
93 * @pinst: padata instance
94 * @padata: object to be parallelized
95 * @cb_cpu: cpu the serialization callback function will run on,
96 * must be in the cpumask of padata.
98 * The parallelization callback function will run with BHs off.
99 * Note: Every object which is parallelized by padata_do_parallel
100 * must be seen by padata_do_serial.
102 int padata_do_parallel(struct padata_instance
*pinst
,
103 struct padata_priv
*padata
, int cb_cpu
)
106 struct padata_queue
*queue
;
107 struct parallel_data
*pd
;
111 pd
= rcu_dereference(pinst
->pd
);
114 if (!(pinst
->flags
& PADATA_INIT
))
118 if ((pinst
->flags
& PADATA_RESET
))
121 if (atomic_read(&pd
->refcnt
) >= MAX_OBJ_NUM
)
125 if (!cpumask_test_cpu(cb_cpu
, pd
->cpumask
))
129 atomic_inc(&pd
->refcnt
);
131 padata
->cb_cpu
= cb_cpu
;
133 if (unlikely(atomic_read(&pd
->seq_nr
) == pd
->max_seq_nr
))
134 atomic_set(&pd
->seq_nr
, -1);
136 padata
->seq_nr
= atomic_inc_return(&pd
->seq_nr
);
138 target_cpu
= padata_cpu_hash(padata
);
139 queue
= per_cpu_ptr(pd
->queue
, target_cpu
);
141 spin_lock(&queue
->parallel
.lock
);
142 list_add_tail(&padata
->list
, &queue
->parallel
.list
);
143 spin_unlock(&queue
->parallel
.lock
);
145 queue_work_on(target_cpu
, pinst
->wq
, &queue
->pwork
);
148 rcu_read_unlock_bh();
152 EXPORT_SYMBOL(padata_do_parallel
);
154 static struct padata_priv
*padata_get_next(struct parallel_data
*pd
)
156 int cpu
, num_cpus
, empty
, calc_seq_nr
;
157 int seq_nr
, next_nr
, overrun
, next_overrun
;
158 struct padata_queue
*queue
, *next_queue
;
159 struct padata_priv
*padata
;
160 struct padata_list
*reorder
;
167 num_cpus
= cpumask_weight(pd
->cpumask
);
169 for_each_cpu(cpu
, pd
->cpumask
) {
170 queue
= per_cpu_ptr(pd
->queue
, cpu
);
171 reorder
= &queue
->reorder
;
174 * Calculate the seq_nr of the object that should be
175 * next in this queue.
178 calc_seq_nr
= (atomic_read(&queue
->num_obj
) * num_cpus
)
181 if (unlikely(calc_seq_nr
> pd
->max_seq_nr
)) {
182 calc_seq_nr
= calc_seq_nr
- pd
->max_seq_nr
- 1;
186 if (!list_empty(&reorder
->list
)) {
187 padata
= list_entry(reorder
->list
.next
,
188 struct padata_priv
, list
);
190 seq_nr
= padata
->seq_nr
;
191 BUG_ON(calc_seq_nr
!= seq_nr
);
193 seq_nr
= calc_seq_nr
;
197 if (next_nr
< 0 || seq_nr
< next_nr
198 || (next_overrun
&& !overrun
)) {
200 next_overrun
= overrun
;
207 if (empty
== num_cpus
)
210 reorder
= &next_queue
->reorder
;
212 if (!list_empty(&reorder
->list
)) {
213 padata
= list_entry(reorder
->list
.next
,
214 struct padata_priv
, list
);
216 if (unlikely(next_overrun
)) {
217 for_each_cpu(cpu
, pd
->cpumask
) {
218 queue
= per_cpu_ptr(pd
->queue
, cpu
);
219 atomic_set(&queue
->num_obj
, 0);
223 spin_lock(&reorder
->lock
);
224 list_del_init(&padata
->list
);
225 atomic_dec(&pd
->reorder_objects
);
226 spin_unlock(&reorder
->lock
);
228 atomic_inc(&next_queue
->num_obj
);
233 if (next_nr
% num_cpus
== next_queue
->cpu_index
) {
234 padata
= ERR_PTR(-ENODATA
);
238 padata
= ERR_PTR(-EINPROGRESS
);
243 static void padata_reorder(struct parallel_data
*pd
)
245 struct padata_priv
*padata
;
246 struct padata_queue
*queue
;
247 struct padata_instance
*pinst
= pd
->pinst
;
250 if (!spin_trylock_bh(&pd
->lock
))
254 padata
= padata_get_next(pd
);
256 if (!padata
|| PTR_ERR(padata
) == -EINPROGRESS
)
259 if (PTR_ERR(padata
) == -ENODATA
) {
260 spin_unlock_bh(&pd
->lock
);
264 queue
= per_cpu_ptr(pd
->queue
, padata
->cb_cpu
);
266 spin_lock(&queue
->serial
.lock
);
267 list_add_tail(&padata
->list
, &queue
->serial
.list
);
268 spin_unlock(&queue
->serial
.lock
);
270 queue_work_on(padata
->cb_cpu
, pinst
->wq
, &queue
->swork
);
273 spin_unlock_bh(&pd
->lock
);
275 if (atomic_read(&pd
->reorder_objects
))
282 static void padata_serial_worker(struct work_struct
*work
)
284 struct padata_queue
*queue
;
285 struct parallel_data
*pd
;
286 LIST_HEAD(local_list
);
289 queue
= container_of(work
, struct padata_queue
, swork
);
292 spin_lock(&queue
->serial
.lock
);
293 list_replace_init(&queue
->serial
.list
, &local_list
);
294 spin_unlock(&queue
->serial
.lock
);
296 while (!list_empty(&local_list
)) {
297 struct padata_priv
*padata
;
299 padata
= list_entry(local_list
.next
,
300 struct padata_priv
, list
);
302 list_del_init(&padata
->list
);
304 padata
->serial(padata
);
305 atomic_dec(&pd
->refcnt
);
311 * padata_do_serial - padata serialization function
313 * @padata: object to be serialized.
315 * padata_do_serial must be called for every parallelized object.
316 * The serialization callback function will run with BHs off.
318 void padata_do_serial(struct padata_priv
*padata
)
321 struct padata_queue
*queue
;
322 struct parallel_data
*pd
;
327 queue
= per_cpu_ptr(pd
->queue
, cpu
);
329 spin_lock(&queue
->reorder
.lock
);
330 atomic_inc(&pd
->reorder_objects
);
331 list_add_tail(&padata
->list
, &queue
->reorder
.list
);
332 spin_unlock(&queue
->reorder
.lock
);
338 EXPORT_SYMBOL(padata_do_serial
);
340 static struct parallel_data
*padata_alloc_pd(struct padata_instance
*pinst
,
341 const struct cpumask
*cpumask
)
343 int cpu
, cpu_index
, num_cpus
;
344 struct padata_queue
*queue
;
345 struct parallel_data
*pd
;
349 pd
= kzalloc(sizeof(struct parallel_data
), GFP_KERNEL
);
353 pd
->queue
= alloc_percpu(struct padata_queue
);
357 if (!alloc_cpumask_var(&pd
->cpumask
, GFP_KERNEL
))
360 for_each_possible_cpu(cpu
) {
361 queue
= per_cpu_ptr(pd
->queue
, cpu
);
365 if (cpumask_test_cpu(cpu
, cpumask
)
366 && cpumask_test_cpu(cpu
, cpu_active_mask
)) {
367 queue
->cpu_index
= cpu_index
;
370 queue
->cpu_index
= -1;
372 INIT_LIST_HEAD(&queue
->reorder
.list
);
373 INIT_LIST_HEAD(&queue
->parallel
.list
);
374 INIT_LIST_HEAD(&queue
->serial
.list
);
375 spin_lock_init(&queue
->reorder
.lock
);
376 spin_lock_init(&queue
->parallel
.lock
);
377 spin_lock_init(&queue
->serial
.lock
);
379 INIT_WORK(&queue
->pwork
, padata_parallel_worker
);
380 INIT_WORK(&queue
->swork
, padata_serial_worker
);
381 atomic_set(&queue
->num_obj
, 0);
384 cpumask_and(pd
->cpumask
, cpumask
, cpu_active_mask
);
386 num_cpus
= cpumask_weight(pd
->cpumask
);
387 pd
->max_seq_nr
= (MAX_SEQ_NR
/ num_cpus
) * num_cpus
- 1;
389 atomic_set(&pd
->seq_nr
, -1);
390 atomic_set(&pd
->reorder_objects
, 0);
391 atomic_set(&pd
->refcnt
, 0);
393 spin_lock_init(&pd
->lock
);
398 free_percpu(pd
->queue
);
405 static void padata_free_pd(struct parallel_data
*pd
)
407 free_cpumask_var(pd
->cpumask
);
408 free_percpu(pd
->queue
);
412 static void padata_replace(struct padata_instance
*pinst
,
413 struct parallel_data
*pd_new
)
415 struct parallel_data
*pd_old
= pinst
->pd
;
417 pinst
->flags
|= PADATA_RESET
;
419 rcu_assign_pointer(pinst
->pd
, pd_new
);
423 while (atomic_read(&pd_old
->refcnt
) != 0)
426 flush_workqueue(pinst
->wq
);
428 padata_free_pd(pd_old
);
430 pinst
->flags
&= ~PADATA_RESET
;
434 * padata_set_cpumask - set the cpumask that padata should use
436 * @pinst: padata instance
437 * @cpumask: the cpumask to use
439 int padata_set_cpumask(struct padata_instance
*pinst
,
440 cpumask_var_t cpumask
)
442 struct parallel_data
*pd
;
447 mutex_lock(&pinst
->lock
);
449 pd
= padata_alloc_pd(pinst
, cpumask
);
455 cpumask_copy(pinst
->cpumask
, cpumask
);
457 padata_replace(pinst
, pd
);
460 mutex_unlock(&pinst
->lock
);
464 EXPORT_SYMBOL(padata_set_cpumask
);
466 static int __padata_add_cpu(struct padata_instance
*pinst
, int cpu
)
468 struct parallel_data
*pd
;
470 if (cpumask_test_cpu(cpu
, cpu_active_mask
)) {
471 pd
= padata_alloc_pd(pinst
, pinst
->cpumask
);
475 padata_replace(pinst
, pd
);
482 * padata_add_cpu - add a cpu to the padata cpumask
484 * @pinst: padata instance
487 int padata_add_cpu(struct padata_instance
*pinst
, int cpu
)
493 mutex_lock(&pinst
->lock
);
495 cpumask_set_cpu(cpu
, pinst
->cpumask
);
496 err
= __padata_add_cpu(pinst
, cpu
);
498 mutex_unlock(&pinst
->lock
);
502 EXPORT_SYMBOL(padata_add_cpu
);
504 static int __padata_remove_cpu(struct padata_instance
*pinst
, int cpu
)
506 struct parallel_data
*pd
;
508 if (cpumask_test_cpu(cpu
, cpu_online_mask
)) {
509 pd
= padata_alloc_pd(pinst
, pinst
->cpumask
);
513 padata_replace(pinst
, pd
);
520 * padata_remove_cpu - remove a cpu from the padata cpumask
522 * @pinst: padata instance
523 * @cpu: cpu to remove
525 int padata_remove_cpu(struct padata_instance
*pinst
, int cpu
)
531 mutex_lock(&pinst
->lock
);
533 cpumask_clear_cpu(cpu
, pinst
->cpumask
);
534 err
= __padata_remove_cpu(pinst
, cpu
);
536 mutex_unlock(&pinst
->lock
);
540 EXPORT_SYMBOL(padata_remove_cpu
);
543 * padata_start - start the parallel processing
545 * @pinst: padata instance to start
547 void padata_start(struct padata_instance
*pinst
)
551 mutex_lock(&pinst
->lock
);
552 pinst
->flags
|= PADATA_INIT
;
553 mutex_unlock(&pinst
->lock
);
555 EXPORT_SYMBOL(padata_start
);
558 * padata_stop - stop the parallel processing
560 * @pinst: padata instance to stop
562 void padata_stop(struct padata_instance
*pinst
)
566 mutex_lock(&pinst
->lock
);
567 pinst
->flags
&= ~PADATA_INIT
;
568 mutex_unlock(&pinst
->lock
);
570 EXPORT_SYMBOL(padata_stop
);
572 static int __cpuinit
padata_cpu_callback(struct notifier_block
*nfb
,
573 unsigned long action
, void *hcpu
)
576 struct padata_instance
*pinst
;
577 int cpu
= (unsigned long)hcpu
;
579 pinst
= container_of(nfb
, struct padata_instance
, cpu_notifier
);
583 case CPU_ONLINE_FROZEN
:
584 if (!cpumask_test_cpu(cpu
, pinst
->cpumask
))
586 mutex_lock(&pinst
->lock
);
587 err
= __padata_add_cpu(pinst
, cpu
);
588 mutex_unlock(&pinst
->lock
);
593 case CPU_DOWN_PREPARE
:
594 case CPU_DOWN_PREPARE_FROZEN
:
595 if (!cpumask_test_cpu(cpu
, pinst
->cpumask
))
597 mutex_lock(&pinst
->lock
);
598 err
= __padata_remove_cpu(pinst
, cpu
);
599 mutex_unlock(&pinst
->lock
);
604 case CPU_UP_CANCELED
:
605 case CPU_UP_CANCELED_FROZEN
:
606 if (!cpumask_test_cpu(cpu
, pinst
->cpumask
))
608 mutex_lock(&pinst
->lock
);
609 __padata_remove_cpu(pinst
, cpu
);
610 mutex_unlock(&pinst
->lock
);
612 case CPU_DOWN_FAILED
:
613 case CPU_DOWN_FAILED_FROZEN
:
614 if (!cpumask_test_cpu(cpu
, pinst
->cpumask
))
616 mutex_lock(&pinst
->lock
);
617 __padata_add_cpu(pinst
, cpu
);
618 mutex_unlock(&pinst
->lock
);
625 * padata_alloc - allocate and initialize a padata instance
627 * @cpumask: cpumask that padata uses for parallelization
628 * @wq: workqueue to use for the allocated padata instance
630 struct padata_instance
*padata_alloc(const struct cpumask
*cpumask
,
631 struct workqueue_struct
*wq
)
634 struct padata_instance
*pinst
;
635 struct parallel_data
*pd
;
637 pinst
= kzalloc(sizeof(struct padata_instance
), GFP_KERNEL
);
641 pd
= padata_alloc_pd(pinst
, cpumask
);
645 rcu_assign_pointer(pinst
->pd
, pd
);
649 cpumask_copy(pinst
->cpumask
, cpumask
);
653 pinst
->cpu_notifier
.notifier_call
= padata_cpu_callback
;
654 pinst
->cpu_notifier
.priority
= 0;
655 err
= register_hotcpu_notifier(&pinst
->cpu_notifier
);
659 mutex_init(&pinst
->lock
);
670 EXPORT_SYMBOL(padata_alloc
);
673 * padata_free - free a padata instance
675 * @ padata_inst: padata instance to free
677 void padata_free(struct padata_instance
*pinst
)
683 while (atomic_read(&pinst
->pd
->refcnt
) != 0)
686 unregister_hotcpu_notifier(&pinst
->cpu_notifier
);
687 padata_free_pd(pinst
->pd
);
690 EXPORT_SYMBOL(padata_free
);