73 static inline struct simple_work_queue *as_simple_work_queue(struct vdo_work_queue *queue)  in as_simple_work_queue()  argument
75 	return ((queue == NULL) ?  in as_simple_work_queue()
76 		NULL : container_of(queue, struct simple_work_queue, common));  in as_simple_work_queue()
79 static inline struct round_robin_work_queue *as_round_robin_work_queue(struct vdo_work_queue *queue)  in as_round_robin_work_queue()  argument
81 	return ((queue == NULL) ?  in as_round_robin_work_queue()
83 		 container_of(queue, struct round_robin_work_queue, common));  in as_round_robin_work_queue()
96 static struct vdo_completion *poll_for_completion(struct simple_work_queue *queue)  in poll_for_completion()  argument
100 	for (i = queue->common.type->max_priority; i >= 0; i--) {  in poll_for_completion()
101 		struct funnel_queue_entry *link = vdo_funnel_queue_poll(queue->priority_lists[i]);  in poll_for_completion()
110 static void enqueue_work_queue_completion(struct simple_work_queue *queue,  in enqueue_work_queue_completion()  argument
115 			    completion, completion->callback, queue, completion->my_queue);  in enqueue_work_queue_completion()
117 		completion->priority = queue->common.type->default_priority;  in enqueue_work_queue_completion()
119 	if (VDO_ASSERT(completion->priority <= queue->common.type->max_priority,  in enqueue_work_queue_completion()
123 	completion->my_queue = &queue->common;  in enqueue_work_queue_completion()
126 	vdo_funnel_queue_put(queue->priority_lists[completion->priority],  in enqueue_work_queue_completion()
146 	if ((atomic_read(&queue->idle) != 1) || (atomic_cmpxchg(&queue->idle, 1, 0) != 1))  in enqueue_work_queue_completion()
150 	wake_up(&queue->waiting_worker_threads);  in enqueue_work_queue_completion()
153 static void run_start_hook(struct simple_work_queue *queue)  in run_start_hook()  argument
155 	if (queue->common.type->start != NULL)  in run_start_hook()
156 		queue->common.type->start(queue->private);  in run_start_hook()
159 static void run_finish_hook(struct simple_work_queue *queue)  in run_finish_hook()  argument
161 	if (queue->common.type->finish != NULL)  in run_finish_hook()
162 		queue->common.type->finish(queue->private);  in run_finish_hook()
174 static struct vdo_completion *wait_for_next_completion(struct simple_work_queue *queue)  in wait_for_next_completion()  argument
180 		prepare_to_wait(&queue->waiting_worker_threads, &wait,  in wait_for_next_completion()
190 		atomic_set(&queue->idle, 1);  in wait_for_next_completion()
193 		completion = poll_for_completion(queue);  in wait_for_next_completion()
211 		completion = poll_for_completion(queue);  in wait_for_next_completion()
216 	finish_wait(&queue->waiting_worker_threads, &wait);  in wait_for_next_completion()
217 	atomic_set(&queue->idle, 0);  in wait_for_next_completion()
222 static void process_completion(struct simple_work_queue *queue,  in process_completion()  argument
225 	if (VDO_ASSERT(completion->my_queue == &queue->common,  in process_completion()
227 		       completion, queue, completion->my_queue) == VDO_SUCCESS)  in process_completion()
233 static void service_work_queue(struct simple_work_queue *queue)  in service_work_queue()  argument
235 	run_start_hook(queue);  in service_work_queue()
238 		struct vdo_completion *completion = poll_for_completion(queue);  in service_work_queue()
241 			completion = wait_for_next_completion(queue);  in service_work_queue()
248 		process_completion(queue, completion);  in service_work_queue()
258 	run_finish_hook(queue);  in service_work_queue()
263 	struct simple_work_queue *queue = ptr;  in work_queue_runner()  local
265 	complete(queue->started);  in work_queue_runner()
266 	service_work_queue(queue);  in work_queue_runner()
272 static void free_simple_work_queue(struct simple_work_queue *queue)  in free_simple_work_queue()  argument
277 		vdo_free_funnel_queue(queue->priority_lists[i]);  in free_simple_work_queue()
278 	vdo_free(queue->common.name);  in free_simple_work_queue()
279 	vdo_free(queue);  in free_simple_work_queue()
282 static void free_round_robin_work_queue(struct round_robin_work_queue *queue)  in free_round_robin_work_queue()  argument
284 	struct simple_work_queue **queue_table = queue->service_queues;  in free_round_robin_work_queue()
285 	unsigned int count = queue->num_service_queues;  in free_round_robin_work_queue()
288 	queue->service_queues = NULL;  in free_round_robin_work_queue()
293 	vdo_free(queue->common.name);  in free_round_robin_work_queue()
294 	vdo_free(queue);  in free_round_robin_work_queue()
297 void vdo_free_work_queue(struct vdo_work_queue *queue)  in vdo_free_work_queue()  argument
299 	if (queue == NULL)  in vdo_free_work_queue()
302 	vdo_finish_work_queue(queue);  in vdo_free_work_queue()
304 	if (queue->round_robin_mode)  in vdo_free_work_queue()
305 		free_round_robin_work_queue(as_round_robin_work_queue(queue));  in vdo_free_work_queue()
307 		free_simple_work_queue(as_simple_work_queue(queue));  in vdo_free_work_queue()
316 	struct simple_work_queue *queue;  in make_simple_work_queue()  local
325 	result = vdo_allocate(1, struct simple_work_queue, "simple work queue", &queue);  in make_simple_work_queue()
329 	queue->private = private;  in make_simple_work_queue()
330 	queue->started = &started;  in make_simple_work_queue()
331 	queue->common.type = type;  in make_simple_work_queue()
332 	queue->common.owner = owner;  in make_simple_work_queue()
333 	init_waitqueue_head(&queue->waiting_worker_threads);  in make_simple_work_queue()
335 	result = vdo_duplicate_string(name, "queue name", &queue->common.name);  in make_simple_work_queue()
337 		vdo_free(queue);  in make_simple_work_queue()
342 		result = vdo_make_funnel_queue(&queue->priority_lists[i]);  in make_simple_work_queue()
344 			free_simple_work_queue(queue);  in make_simple_work_queue()
349 	thread = kthread_run(work_queue_runner, queue, "%s:%s", thread_name_prefix,  in make_simple_work_queue()
350 			     queue->common.name);  in make_simple_work_queue()
352 		free_simple_work_queue(queue);  in make_simple_work_queue()
356 	queue->thread = thread;  in make_simple_work_queue()
368 	*queue_ptr = queue;  in make_simple_work_queue()
392 	struct round_robin_work_queue *queue;  in vdo_make_work_queue()  local
409 			      &queue);  in vdo_make_work_queue()
414 			      "subordinate work queues", &queue->service_queues);  in vdo_make_work_queue()
416 		vdo_free(queue);  in vdo_make_work_queue()
420 	queue->num_service_queues = thread_count;  in vdo_make_work_queue()
421 	queue->common.round_robin_mode = true;  in vdo_make_work_queue()
422 	queue->common.owner = owner;  in vdo_make_work_queue()
424 	result = vdo_duplicate_string(name, "queue name", &queue->common.name);  in vdo_make_work_queue()
426 		vdo_free(queue->service_queues);  in vdo_make_work_queue()
427 		vdo_free(queue);  in vdo_make_work_queue()
431 	*queue_ptr = &queue->common;  in vdo_make_work_queue()
438 						context, type, &queue->service_queues[i]);  in vdo_make_work_queue()
440 			queue->num_service_queues = i;  in vdo_make_work_queue()
450 static void finish_simple_work_queue(struct simple_work_queue *queue)  in finish_simple_work_queue()  argument
452 	if (queue->thread == NULL)  in finish_simple_work_queue()
456 	kthread_stop(queue->thread);  in finish_simple_work_queue()
457 	queue->thread = NULL;  in finish_simple_work_queue()
460 static void finish_round_robin_work_queue(struct round_robin_work_queue *queue)  in finish_round_robin_work_queue()  argument
462 	struct simple_work_queue **queue_table = queue->service_queues;  in finish_round_robin_work_queue()
463 	unsigned int count = queue->num_service_queues;  in finish_round_robin_work_queue()
471 void vdo_finish_work_queue(struct vdo_work_queue *queue)  in vdo_finish_work_queue()  argument
473 	if (queue == NULL)  in vdo_finish_work_queue()
476 	if (queue->round_robin_mode)  in vdo_finish_work_queue()
477 		finish_round_robin_work_queue(as_round_robin_work_queue(queue));  in vdo_finish_work_queue()
479 		finish_simple_work_queue(as_simple_work_queue(queue));  in vdo_finish_work_queue()
484 static void dump_simple_work_queue(struct simple_work_queue *queue)  in dump_simple_work_queue()  argument
489 	if (queue->thread != NULL) {  in dump_simple_work_queue()
490 		task_state_report = task_state_to_char(queue->thread);  in dump_simple_work_queue()
491 		thread_status = atomic_read(&queue->idle) ? "idle" : "running";  in dump_simple_work_queue()
494 	vdo_log_info("workQ %px (%s) %s (%c)", &queue->common, queue->common.name,  in dump_simple_work_queue()
505 void vdo_dump_work_queue(struct vdo_work_queue *queue)  in vdo_dump_work_queue()  argument
507 	if (queue->round_robin_mode) {  in vdo_dump_work_queue()
508 		struct round_robin_work_queue *round_robin = as_round_robin_work_queue(queue);  in vdo_dump_work_queue()
514 		dump_simple_work_queue(as_simple_work_queue(queue));  in vdo_dump_work_queue()
562 void vdo_enqueue_work_queue(struct vdo_work_queue *queue,  in vdo_enqueue_work_queue()  argument
571 	if (!queue->round_robin_mode) {  in vdo_enqueue_work_queue()
572 		simple_queue = as_simple_work_queue(queue);  in vdo_enqueue_work_queue()
574 		struct round_robin_work_queue *round_robin = as_round_robin_work_queue(queue);  in vdo_enqueue_work_queue()
618 	struct simple_work_queue *queue = get_current_thread_work_queue();  in vdo_get_current_work_queue()  local
620 	return (queue == NULL) ? NULL : &queue->common;  in vdo_get_current_work_queue()
623 struct vdo_thread *vdo_get_work_queue_owner(struct vdo_work_queue *queue)  in vdo_get_work_queue_owner()  argument
625 	return queue->owner;  in vdo_get_work_queue_owner()
635 	struct simple_work_queue *queue = get_current_thread_work_queue();  in vdo_get_work_queue_private_data()  local
637 	return (queue != NULL) ? queue->private : NULL;  in vdo_get_work_queue_private_data()
640 bool vdo_work_queue_type_is(struct vdo_work_queue *queue,  in vdo_work_queue_type_is()  argument
643 	return (queue->type == type);  in vdo_work_queue_type_is()