diff --git a/mse_core_main.c b/mse_core_main.c index bd6f195..a863251 100644 --- a/mse_core_main.c +++ b/mse_core_main.c @@ -70,8 +70,7 @@ #include #include #include -#include -#include +#include #include #include #include @@ -280,14 +279,6 @@ struct mse_wait_packet { struct list_head list; }; -/** @brief workqueue */ -struct mse_workqueue { - /** @brief kthread worker to queue the work */ - struct kthread_worker wrk; - /** @brief task of kthread worker */ - struct task_struct *tsk; -}; - /** @brief instance by related adapter */ struct mse_instance { /** @brief wait for streaming stop */ @@ -315,32 +306,32 @@ struct mse_instance { enum MSE_PACKETIZER packetizer_id; /** @brief streaming queue */ - struct kthread_work wk_stream; + struct work_struct wk_stream; /** @brief paketize queue */ - struct kthread_work wk_packetize; + struct work_struct wk_packetize; /** @brief depaketize queue */ - struct kthread_work wk_depacketize; + struct work_struct wk_depacketize; /** @brief callback queue */ - struct kthread_work wk_callback; + struct work_struct wk_callback; /** @brief timestamp queue */ - struct kthread_work wk_timestamp; + struct work_struct wk_timestamp; /** @brief crf send queue */ - struct kthread_work wk_crf_send; + struct work_struct wk_crf_send; /** @brief crf receive queue */ - struct kthread_work wk_crf_receive; + struct work_struct wk_crf_receive; /** @brief start transmission queue */ - struct kthread_work wk_start_trans; + struct work_struct wk_start_trans; /** @brief stop streaming queue */ - struct kthread_work wk_stop_streaming; + struct work_struct wk_stop_streaming; /** @brief stream workqueue */ - struct mse_workqueue wq_stream; + struct workqueue_struct *wq_stream; /** @brief packet workqueue */ - struct mse_workqueue wq_packet; + struct workqueue_struct *wq_packet; /** @brief timestamp workqueue */ - struct mse_workqueue wq_tstamp; + struct workqueue_struct *wq_tstamp; /** @brief crf packet workqueue */ - struct mse_workqueue wq_crf_packet; + struct workqueue_struct *wq_crf_packet; /** @brief wait queue for streaming */ wait_queue_head_t wait_wk_stream; @@ -528,10 +519,6 @@ static struct mse_device *mse; static int major; module_param(major, int, 0440); -static int rt_prio; -module_param(rt_prio, int, 0660); -MODULE_PARM_DESC(rt_prio, "apply RT priority to worker threads (1-99) or do NOT apply RT priority (0)"); - /* * function prototypes */ @@ -1127,11 +1114,6 @@ static void mse_free_all_trans_buffers(struct mse_instance *instance, int size) atomic_set(&instance->trans_buf_cnt, 0); } -#define mse_queue_work(_q, _wrk) kthread_queue_work(&(_q).wrk, _wrk) -#define mse_flush_work(_wrk) kthread_flush_work(_wrk) -#define mse_flush_workqueue(_q) kthread_flush_worker(&(_q).wrk) -#define mse_destroy_workqueue(_q) kthread_stop((_q).tsk) - static void mse_work_stream_common(struct mse_instance *instance) { int index_network; @@ -1185,13 +1167,17 @@ static void mse_work_stream_common(struct mse_instance *instance) break; } - /* process depacketize */ - mse_queue_work(instance->wq_packet, &instance->wk_depacketize); + /* if NOT work queued, then queue it */ + if (!work_busy(&instance->wk_depacketize)) + queue_work(instance->wq_packet, + &instance->wk_depacketize); } - /* process depacketize to process last data */ + /* if NOT work queued, then queue it to process last data */ if (mse_state_test(instance, MSE_STATE_STOPPING)) - mse_queue_work(instance->wq_packet, &instance->wk_depacketize); + if (!work_busy(&instance->wk_depacketize)) + queue_work(instance->wq_packet, + &instance->wk_depacketize); } write_lock_irqsave(&instance->lock_stream, flags); @@ -1247,7 +1233,7 @@ static void mse_work_stream_mpeg2ts_tx(struct mse_instance *instance) mse_debug("END\n"); } -static void mse_work_stream(struct kthread_work *work) +static void mse_work_stream(struct work_struct *work) { struct mse_instance *instance; @@ -1854,7 +1840,7 @@ static int media_clock_recovery(struct mse_instance *instance) return out; } -static void mse_work_timestamp(struct kthread_work *work) +static void mse_work_timestamp(struct work_struct *work) { struct mse_instance *instance; int captured; @@ -2125,7 +2111,8 @@ static void mse_work_packetize_common(struct mse_instance *instance) read_lock_irqsave(&instance->lock_stream, flags); if (!instance->f_streaming) - mse_queue_work(instance->wq_stream, &instance->wk_stream); + queue_work(instance->wq_stream, + &instance->wk_stream); read_unlock_irqrestore(&instance->lock_stream, flags); @@ -2155,7 +2142,7 @@ static void mse_work_packetize_common(struct mse_instance *instance) /* start workqueue for streaming */ read_lock_irqsave(&instance->lock_stream, flags); if (!instance->f_streaming && ret > 0) - mse_queue_work(instance->wq_stream, &instance->wk_stream); + queue_work(instance->wq_stream, &instance->wk_stream); read_unlock_irqrestore(&instance->lock_stream, flags); } @@ -2171,7 +2158,8 @@ static void mse_work_packetize_common(struct mse_instance *instance) mse_debug("discard %zu byte before stopping\n", buf->buffer_size - buf->work_length); - mse_queue_work(instance->wq_packet, &instance->wk_stop_streaming); + queue_work(instance->wq_packet, + &instance->wk_stop_streaming); } else { mse_err("short of data\n"); @@ -2198,7 +2186,8 @@ static void mse_work_packetize_common(struct mse_instance *instance) /* state is STOPPING */ if (mse_state_test(instance, MSE_STATE_STOPPING)) { - mse_queue_work(instance->wq_packet, &instance->wk_stop_streaming); + queue_work(instance->wq_packet, + &instance->wk_stop_streaming); } else { write_lock_irqsave(&instance->lock_state, flags); /* if state is EXECUTE, change to IDLE */ @@ -2211,13 +2200,13 @@ static void mse_work_packetize_common(struct mse_instance *instance) } if (instance->f_continue) { - mse_queue_work(instance->wq_packet, &instance->wk_packetize); + queue_work(instance->wq_packet, &instance->wk_packetize); } else { if (!instance->timer_interval) atomic_inc(&instance->done_buf_cnt); if (atomic_read(&instance->done_buf_cnt) > 0) - mse_queue_work(instance->wq_packet, &instance->wk_callback); + queue_work(instance->wq_packet, &instance->wk_callback); } } @@ -2576,7 +2565,7 @@ static void mse_control_wait_packet(struct mse_instance *instance, read_lock_irqsave(&instance->lock_stream, flags); /* start workqueue for streaming */ if (!instance->f_streaming) - mse_queue_work(instance->wq_stream, &instance->wk_stream); + queue_work(instance->wq_stream, &instance->wk_stream); read_unlock_irqrestore(&instance->lock_stream, flags); return; @@ -2593,7 +2582,7 @@ static void mse_control_wait_packet(struct mse_instance *instance, /* start workqueue for streaming */ if (!instance->f_streaming) - mse_queue_work(instance->wq_stream, &instance->wk_stream); + queue_work(instance->wq_stream, &instance->wk_stream); } else { /* Update wait packet queue when start timer */ mse_update_wait_packet(instance, timestamp); @@ -2613,7 +2602,7 @@ static void mse_control_wait_packet(struct mse_instance *instance, INIT_LIST_HEAD(&instance->wait_packet_list); mse_packet_ctrl_release_all_wait(instance->packet_buffer); if (!instance->f_streaming) - mse_queue_work(instance->wq_stream, &instance->wk_stream); + queue_work(instance->wq_stream, &instance->wk_stream); } else { instance->f_timer_started = true; @@ -2679,7 +2668,7 @@ static void mse_work_packetize_mpeg2ts_tx(struct mse_instance *instance) false, &piece_length); - mse_queue_work(instance->wq_packet, &instance->wk_stop_streaming); + queue_work(instance->wq_packet, &instance->wk_stop_streaming); } return; /* skip work */ } @@ -2691,7 +2680,7 @@ static void mse_work_packetize_mpeg2ts_tx(struct mse_instance *instance) if (trans_size <= 0) { /* no data to process */ list_move_tail(&buf->list, &instance->done_buf_list); - mse_queue_work(instance->wq_packet, &instance->wk_callback); + queue_work(instance->wq_packet, &instance->wk_callback); return; } @@ -2720,7 +2709,8 @@ static void mse_work_packetize_mpeg2ts_tx(struct mse_instance *instance) read_lock_irqsave(&instance->lock_stream, flags); if (!instance->f_streaming) - mse_queue_work(instance->wq_stream, &instance->wk_stream); + queue_work(instance->wq_stream, + &instance->wk_stream); read_unlock_irqrestore(&instance->lock_stream, flags); @@ -2755,20 +2745,20 @@ static void mse_work_packetize_mpeg2ts_tx(struct mse_instance *instance) } if (instance->f_continue) { - mse_queue_work(instance->wq_packet, &instance->wk_packetize); + queue_work(instance->wq_packet, &instance->wk_packetize); } else { list_move_tail(&buf->list, &instance->done_buf_list); - mse_queue_work(instance->wq_packet, &instance->wk_callback); + queue_work(instance->wq_packet, &instance->wk_callback); if (!list_empty(&instance->proc_buf_list) || mse_state_test(instance, MSE_STATE_STOPPING)) { - mse_queue_work(instance->wq_packet, &instance->wk_packetize); + queue_work(instance->wq_packet, &instance->wk_packetize); } } } -static void mse_work_packetize(struct kthread_work *work) +static void mse_work_packetize(struct work_struct *work) { struct mse_instance *instance; @@ -2826,14 +2816,15 @@ static void mse_work_depacketize_common(struct mse_instance *instance) if (!buf) { /* state is STOPPING */ if (mse_state_test(instance, MSE_STATE_STOPPING)) - mse_queue_work(instance->wq_packet, &instance->wk_stop_streaming); + queue_work(instance->wq_packet, + &instance->wk_stop_streaming); return; } read_lock_irqsave(&instance->lock_stream, flags); if (!instance->f_streaming) - mse_queue_work(instance->wq_stream, &instance->wk_stream); + queue_work(instance->wq_stream, &instance->wk_stream); read_unlock_irqrestore(&instance->lock_stream, flags); instance->f_depacketizing = true; @@ -2967,13 +2958,14 @@ static void mse_work_depacketize_common(struct mse_instance *instance) if (atomic_read(&instance->done_buf_cnt) > 0 && has_valid_data) - mse_queue_work(instance->wq_packet, &instance->wk_callback); + queue_work(instance->wq_packet, &instance->wk_callback); /* state is STOPPING */ if (mse_state_test(instance, MSE_STATE_STOPPING)) { if (instance->ptp_timer_handle || !atomic_read(&instance->done_buf_cnt)) - mse_queue_work(instance->wq_packet, &instance->wk_stop_streaming); + queue_work(instance->wq_packet, + &instance->wk_stop_streaming); if (atomic_inc_return(&instance->done_buf_cnt) != 1) atomic_dec(&instance->done_buf_cnt); @@ -3167,7 +3159,7 @@ static void mse_work_depacketize_video_rx(struct mse_instance *instance) if (wait_buf) { /* Move from wait_buf_list to done_buf_list */ list_move_tail(&wait_buf->list, &instance->done_buf_list); - mse_queue_work(instance->wq_packet, &instance->wk_callback); + queue_work(instance->wq_packet, &instance->wk_callback); } } @@ -3178,7 +3170,7 @@ static void mse_work_depacketize_video_rx(struct mse_instance *instance) read_lock_irqsave(&instance->lock_stream, flags); if (!instance->f_streaming) - mse_queue_work(instance->wq_stream, &instance->wk_stream); + queue_work(instance->wq_stream, &instance->wk_stream); read_unlock_irqrestore(&instance->lock_stream, flags); instance->f_depacketizing = true; @@ -3192,14 +3184,14 @@ static void mse_work_depacketize_video_rx(struct mse_instance *instance) * wk_callback */ if (!list_empty(&instance->done_buf_list)) - mse_queue_work(instance->wq_packet, &instance->wk_callback); + queue_work(instance->wq_packet, &instance->wk_callback); /* state is NOT EXECUTE */ if (!mse_state_test(instance, MSE_STATE_EXECUTE)) instance->f_depacketizing = false; } -static void mse_work_depacketize(struct kthread_work *work) +static void mse_work_depacketize(struct work_struct *work) { struct mse_instance *instance; @@ -3231,7 +3223,8 @@ static void mse_work_callback_common(struct mse_instance *instance) if (!buf) { /* state is STOPPING */ if (mse_state_test(instance, MSE_STATE_STOPPING)) - mse_queue_work(instance->wq_packet, &instance->wk_stop_streaming); + queue_work(instance->wq_packet, + &instance->wk_stop_streaming); return; /* skip work */ } @@ -3345,7 +3338,8 @@ static void mse_work_callback_common(struct mse_instance *instance) /* state is STOPPING */ if (mse_state_test(instance, MSE_STATE_STOPPING)) - mse_queue_work(instance->wq_packet, &instance->wk_stop_streaming); + queue_work(instance->wq_packet, + &instance->wk_stop_streaming); } /* complete callback */ @@ -3360,19 +3354,22 @@ static void mse_work_callback_common(struct mse_instance *instance) if (mse_is_buffer_empty(instance)) { /* state is STOPPING */ if (mse_state_test_nolock(instance, MSE_STATE_STOPPING)) { - mse_queue_work(instance->wq_packet, &instance->wk_stop_streaming); + queue_work(instance->wq_packet, + &instance->wk_stop_streaming); } else { /* if state is EXECUTE, change to IDLE */ mse_state_change_if(instance, MSE_STATE_IDLE, MSE_STATE_EXECUTE); } } else { - /* queue it */ - if (instance->tx) - mse_queue_work(instance->wq_packet, &instance->wk_packetize); + /* if NOT work queued, then queue it */ + if (instance->tx && !work_busy(&instance->wk_packetize)) + queue_work(instance->wq_packet, + &instance->wk_packetize); - if (!instance->tx) - mse_queue_work(instance->wq_packet, &instance->wk_depacketize); + if (!instance->tx && !work_busy(&instance->wk_depacketize)) + queue_work(instance->wq_packet, + &instance->wk_depacketize); } write_unlock_irqrestore(&instance->lock_state, flags); @@ -3411,7 +3408,7 @@ static void mse_work_callback_video_rx(struct mse_instance *instance) /* state is STOPPING */ if (mse_state_test_nolock(instance, MSE_STATE_STOPPING)) { - mse_queue_work(instance->wq_packet, &instance->wk_stop_streaming); + queue_work(instance->wq_packet, &instance->wk_stop_streaming); } else { /* if state is EXECUTE, change to IDLE */ mse_state_change_if(instance, MSE_STATE_IDLE, MSE_STATE_EXECUTE); @@ -3447,14 +3444,14 @@ static void mse_work_callback_mpeg2ts_tx(struct mse_instance *instance) /* buffer is NOT empty, so wait next callback queued */ if (!mse_is_buffer_empty(instance)) { - mse_queue_work(instance->wq_packet, &instance->wk_packetize); + queue_work(instance->wq_packet, &instance->wk_packetize); write_unlock_irqrestore(&instance->lock_state, flags); return; } /* state is STOPPING */ if (mse_state_test_nolock(instance, MSE_STATE_STOPPING)) { - mse_queue_work(instance->wq_packet, &instance->wk_stop_streaming); + queue_work(instance->wq_packet, &instance->wk_stop_streaming); } else { /* if state is EXECUTE, change to IDLE */ mse_state_change_if(instance, MSE_STATE_IDLE, MSE_STATE_EXECUTE); @@ -3463,7 +3460,7 @@ static void mse_work_callback_mpeg2ts_tx(struct mse_instance *instance) write_unlock_irqrestore(&instance->lock_state, flags); } -static void mse_work_callback(struct kthread_work *work) +static void mse_work_callback(struct work_struct *work) { struct mse_instance *instance; @@ -3498,14 +3495,14 @@ static void mse_stop_streaming_audio(struct mse_instance *instance) if (ret) mse_err("failed cancel() ret=%d\n", ret); - mse_flush_work(&instance->wk_crf_receive); + flush_work(&instance->wk_crf_receive); } else if (crf_type == MSE_CRF_TYPE_TX && instance->f_crf_sending) { - mse_flush_work(&instance->wk_crf_send); + flush_work(&instance->wk_crf_send); } if (instance->f_work_timestamp) - mse_flush_work(&instance->wk_timestamp); + flush_work(&instance->wk_timestamp); } static void mse_stop_streaming_common(struct mse_instance *instance) @@ -3580,13 +3577,13 @@ static void mse_work_stop_streaming_common(struct mse_instance *instance) write_unlock_irqrestore(&instance->lock_state, flags); if (instance->tx) { - mse_queue_work(instance->wq_packet, &instance->wk_start_trans); + queue_work(instance->wq_packet, &instance->wk_start_trans); } else { ret = network->cancel(instance->index_network); if (ret) mse_err("failed network adapter cancel() => %d\n", ret); - mse_queue_work(instance->wq_packet, &instance->wk_depacketize); + queue_work(instance->wq_packet, &instance->wk_depacketize); } } @@ -3635,10 +3632,10 @@ static void mse_work_stop_streaming_mpeg2ts_tx(struct mse_instance *instance) mse_state_change(instance, MSE_STATE_STOPPING); write_unlock_irqrestore(&instance->lock_state, flags); - mse_queue_work(instance->wq_packet, &instance->wk_packetize); + queue_work(instance->wq_packet, &instance->wk_packetize); } -static void mse_work_stop_streaming(struct kthread_work *work) +static void mse_work_stop_streaming(struct work_struct *work) { struct mse_instance *instance; @@ -3674,7 +3671,7 @@ static enum hrtimer_restart mse_timer_callback(struct hrtimer *arg) hrtimer_add_expires_ns(&instance->timer, instance->timer_interval); /* start workqueue for completion */ - mse_queue_work(instance->wq_packet, &instance->wk_callback); + queue_work(instance->wq_packet, &instance->wk_callback); return HRTIMER_RESTART; } @@ -3695,7 +3692,7 @@ static u32 mse_ptp_timer_callback_common(struct mse_instance *instance) if (IS_MSE_TYPE_AUDIO(instance->media->type)) atomic_inc(&instance->done_buf_cnt); - mse_queue_work(instance->wq_packet, &instance->wk_callback); + queue_work(instance->wq_packet, &instance->wk_callback); return (u32)(expire_next - expire_prev); } @@ -3737,7 +3734,7 @@ static u32 mse_ptp_timer_callback_video_rx(struct mse_instance *instance) /* Output buffer */ list_move_tail(&buf->list, &instance->done_buf_list); - mse_queue_work(instance->wq_packet, &instance->wk_callback); + queue_work(instance->wq_packet, &instance->wk_callback); } /* If not found next timestamp, timer stop */ @@ -3789,7 +3786,7 @@ static u32 mse_ptp_timer_callback_mpeg2ts_tx(struct mse_instance *instance) /* start workqueue for streaming */ if (!instance->f_streaming) - mse_queue_work(instance->wq_stream, &instance->wk_stream); + queue_work(instance->wq_stream, &instance->wk_stream); /* If not found next timestamp, timer stop */ if (expire_next == expire_prev) @@ -3818,7 +3815,7 @@ static u32 mse_ptp_timer_callback(void *arg) return mse_ptp_timer_callback_common(instance); } -static void mse_work_crf_send(struct kthread_work *work) +static void mse_work_crf_send(struct work_struct *work) { struct mse_instance *instance; int err, tsize, size, i; @@ -3881,7 +3878,7 @@ static void mse_work_crf_send(struct kthread_work *work) instance->f_crf_sending = false; } -static void mse_work_crf_receive(struct kthread_work *work) +static void mse_work_crf_receive(struct work_struct *work) { struct mse_instance *instance; struct mse_audio_info audio_info; @@ -3946,7 +3943,7 @@ static enum hrtimer_restart mse_crf_callback(struct hrtimer *arg) /* start workqueue for send */ if (!instance->f_crf_sending) { instance->f_crf_sending = true; - mse_queue_work(instance->wq_crf_packet, &instance->wk_crf_send); + queue_work(instance->wq_crf_packet, &instance->wk_crf_send); } return HRTIMER_RESTART; @@ -3974,10 +3971,10 @@ static enum hrtimer_restart mse_timestamp_collect_callback(struct hrtimer *arg) return HRTIMER_RESTART; instance->f_work_timestamp = true; - mse_queue_work(instance->wq_tstamp, &instance->wk_timestamp); + queue_work(instance->wq_tstamp, &instance->wk_timestamp); if (instance->f_wait_start_transmission) { - mse_queue_work(instance->wq_packet, &instance->wk_start_trans); + queue_work(instance->wq_packet, &instance->wk_start_trans); instance->f_wait_start_transmission = false; } @@ -4099,7 +4096,8 @@ static void mse_start_streaming_audio(struct mse_instance *instance, u64 now) /* receive clock using CRF */ if (instance->crf_type == MSE_CRF_TYPE_RX) { - mse_queue_work(instance->wq_crf_packet, &instance->wk_crf_receive); + queue_work(instance->wq_crf_packet, + &instance->wk_crf_receive); } } @@ -4203,7 +4201,8 @@ static void mse_work_start_transmission_common(struct mse_instance *instance) spin_unlock_irqrestore(&instance->lock_buf_list, flags); /* state is STOPPING */ if (mse_state_test(instance, MSE_STATE_STOPPING)) { - mse_queue_work(instance->wq_packet, &instance->wk_stop_streaming); + queue_work(instance->wq_packet, + &instance->wk_stop_streaming); } return; @@ -4211,7 +4210,8 @@ static void mse_work_start_transmission_common(struct mse_instance *instance) if (mse_state_test(instance, MSE_STATE_STOPPING)) { spin_unlock_irqrestore(&instance->lock_buf_list, flags); - mse_queue_work(instance->wq_packet, &instance->wk_stop_streaming); + queue_work(instance->wq_packet, + &instance->wk_stop_streaming); return; } } @@ -4274,13 +4274,13 @@ static void mse_work_start_transmission_common(struct mse_instance *instance) } /* start workqueue for packetize */ - mse_queue_work(instance->wq_packet, &instance->wk_packetize); + queue_work(instance->wq_packet, &instance->wk_packetize); } else { if (buf->buffer) memset(buf->buffer, 0, buf->buffer_size); /* start workqueue for depacketize */ - mse_queue_work(instance->wq_packet, &instance->wk_depacketize); + queue_work(instance->wq_packet, &instance->wk_depacketize); } } @@ -4298,7 +4298,7 @@ static void mse_work_start_transmission_video_rx(struct mse_instance *instance) spin_unlock_irqrestore(&instance->lock_buf_list, flags); /* start workqueue for depacketize */ - mse_queue_work(instance->wq_packet, &instance->wk_depacketize); + queue_work(instance->wq_packet, &instance->wk_depacketize); } static void mse_work_start_transmission_mpeg2ts_tx(struct mse_instance *instance) @@ -4315,10 +4315,10 @@ static void mse_work_start_transmission_mpeg2ts_tx(struct mse_instance *instance spin_unlock_irqrestore(&instance->lock_buf_list, flags); /* start workqueue for packetize */ - mse_queue_work(instance->wq_packet, &instance->wk_packetize); + queue_work(instance->wq_packet, &instance->wk_packetize); } -static void mse_work_start_transmission(struct kthread_work *work) +static void mse_work_start_transmission(struct work_struct *work) { struct mse_instance *instance; @@ -5256,60 +5256,37 @@ static void mse_exit_kernel_resource(struct mse_instance *instance, { /* flush workqueue */ - if (instance->wq_crf_packet.tsk) - mse_flush_workqueue(instance->wq_crf_packet); + if (instance->wq_crf_packet) + flush_workqueue(instance->wq_crf_packet); - if (instance->wq_tstamp.tsk) - mse_flush_workqueue(instance->wq_tstamp); + if (instance->wq_tstamp) + flush_workqueue(instance->wq_tstamp); - if (instance->wq_packet.tsk) - mse_flush_workqueue(instance->wq_packet); + if (instance->wq_packet) + flush_workqueue(instance->wq_packet); - if (instance->wq_stream.tsk) - mse_flush_workqueue(instance->wq_stream); + if (instance->wq_stream) + flush_workqueue(instance->wq_stream); /* destroy workqueue */ - if (instance->wq_crf_packet.tsk) - mse_destroy_workqueue(instance->wq_crf_packet); + if (instance->wq_crf_packet) + destroy_workqueue(instance->wq_crf_packet); - if (instance->wq_tstamp.tsk) - mse_destroy_workqueue(instance->wq_tstamp); + if (instance->wq_tstamp) + destroy_workqueue(instance->wq_tstamp); - if (instance->wq_packet.tsk) - mse_destroy_workqueue(instance->wq_packet); + if (instance->wq_packet) + destroy_workqueue(instance->wq_packet); - if (instance->wq_stream.tsk) - mse_destroy_workqueue(instance->wq_stream); -} - -static int mse_create_workqueue(struct mse_workqueue *wrk, const char *name) -{ - kthread_init_worker(&wrk->wrk); - wrk->tsk = kthread_run(kthread_worker_fn, &wrk->wrk, "%s", name); - if (IS_ERR(wrk->tsk)) { - int err = PTR_ERR(wrk->tsk); - /* set to NULL for easier ptr check in cleanup path */ - wrk->tsk = NULL; - return err; - } - - /* rt priority needed? */ - if (rt_prio > 0) { - struct sched_param param = { 0 }; - - if (rt_prio > (MAX_RT_PRIO - 1)) { - mse_warn("limit rt_prio val %d to maximum %d\n", rt_prio, MAX_RT_PRIO - 1); - rt_prio = MAX_RT_PRIO - 1; - } - param.sched_priority = rt_prio; - sched_setscheduler(wrk->tsk, SCHED_FIFO, ¶m); - } - return 0; + if (instance->wq_stream) + destroy_workqueue(instance->wq_stream); } static int mse_init_kernel_resource(struct mse_instance *instance, struct mse_adapter *adapter) { + struct workqueue_struct *wq_work; + init_completion(&instance->completion_stop); complete(&instance->completion_stop); atomic_set(&instance->trans_buf_cnt, 0); @@ -5327,25 +5304,34 @@ static int mse_init_kernel_resource(struct mse_instance *instance, sema_init(&instance->sem_stopping, 1); /* init work queue */ - kthread_init_work(&instance->wk_packetize, mse_work_packetize); - kthread_init_work(&instance->wk_depacketize, mse_work_depacketize); - kthread_init_work(&instance->wk_callback, mse_work_callback); - kthread_init_work(&instance->wk_stream, mse_work_stream); - kthread_init_work(&instance->wk_crf_send, mse_work_crf_send); - kthread_init_work(&instance->wk_crf_receive, mse_work_crf_receive); - kthread_init_work(&instance->wk_timestamp, mse_work_timestamp); - kthread_init_work(&instance->wk_start_trans, mse_work_start_transmission); - kthread_init_work(&instance->wk_stop_streaming, mse_work_stop_streaming); - - if (mse_create_workqueue(&instance->wq_stream, "mse_streamq") < 0) - goto error_create_wq; - - if (mse_create_workqueue(&instance->wq_packet, "mse_packetq") < 0) - goto error_create_wq; + INIT_WORK(&instance->wk_packetize, mse_work_packetize); + INIT_WORK(&instance->wk_depacketize, mse_work_depacketize); + INIT_WORK(&instance->wk_callback, mse_work_callback); + INIT_WORK(&instance->wk_stream, mse_work_stream); + INIT_WORK(&instance->wk_crf_send, mse_work_crf_send); + INIT_WORK(&instance->wk_crf_receive, mse_work_crf_receive); + INIT_WORK(&instance->wk_timestamp, mse_work_timestamp); + INIT_WORK(&instance->wk_start_trans, mse_work_start_transmission); + INIT_WORK(&instance->wk_stop_streaming, mse_work_stop_streaming); + + wq_work = create_singlethread_workqueue("mse_streamq"); + if (!wq_work) + goto error_create_singlethread_wq; + + instance->wq_stream = wq_work; + + wq_work = create_singlethread_workqueue("mse_packetq"); + if (!wq_work) + goto error_create_singlethread_wq; + + instance->wq_packet = wq_work; /* for timestamp */ - if (mse_create_workqueue(&instance->wq_tstamp, "mse_tstampq") < 0) - goto error_create_wq; + wq_work = create_singlethread_workqueue("mse_tstampq"); + if (!wq_work) + goto error_create_singlethread_wq; + + instance->wq_tstamp = wq_work; hrtimer_init(&instance->timer, CLOCK_MONOTONIC, HRTIMER_MODE_REL); instance->timer_interval = 0; @@ -5359,8 +5345,11 @@ static int mse_init_kernel_resource(struct mse_instance *instance, &mse_timestamp_collect_callback; /* for crf */ - if (mse_create_workqueue(&instance->wq_crf_packet, "mse_crfpacketq") < 0) - goto error_create_wq; + wq_work = create_singlethread_workqueue("mse_crfpacketq"); + if (!wq_work) + goto error_create_singlethread_wq; + + instance->wq_crf_packet = wq_work; hrtimer_init(&instance->crf_timer, CLOCK_MONOTONIC, HRTIMER_MODE_REL); @@ -5370,7 +5359,7 @@ static int mse_init_kernel_resource(struct mse_instance *instance, return 0; -error_create_wq: +error_create_singlethread_wq: mse_exit_kernel_resource(instance, adapter); return -ENOMEM; @@ -5893,7 +5882,7 @@ int mse_stop_streaming(int index) /* state is RUNNABLE */ if (mse_state_test(instance, MSE_STATE_RUNNABLE)) { down(&instance->sem_stopping); - mse_queue_work(instance->wq_packet, &instance->wk_stop_streaming); + queue_work(instance->wq_packet, &instance->wk_stop_streaming); } return 0; @@ -5992,7 +5981,7 @@ int mse_start_transmission(int index, atomic_inc(&instance->trans_buf_cnt); spin_unlock_irqrestore(&instance->lock_buf_list, flags); - mse_queue_work(instance->wq_packet, &instance->wk_start_trans); + queue_work(instance->wq_packet, &instance->wk_start_trans); } write_unlock_irqrestore(&instance->lock_state, flags2);