Re: [PATCH for-next 1/2] RDMA/hns: Add the workqueue framework for flush cqe handler

["Liuyixian (Eason)" <liuyixian@huawei.com>]

On 2019/11/7 4:40, Jason Gunthorpe wrote:
> On Mon, Oct 28, 2019 at 05:45:44PM +0800, Yixian Liu wrote:
>> @@ -1998,6 +2000,17 @@ static int hns_roce_v2_init(struct hns_roce_dev *hr_dev)
>>  		}
>>  	}
>>  
>> +	snprintf(workq_name, HNS_ROCE_WORKQ_NAME_LEN - 1,
>> +		 "hns_roce_%d_flush_wq", device_id);
>> +	device_id++;
>> +
>> +	hr_dev->flush_workq = alloc_workqueue(workq_name, WQ_HIGHPRI, 0);
>> +	if (!hr_dev->flush_workq) {
> 
> Why is this so time critical?

Hi Jason,

I am not quite sure whether you concerned with the flag "WQ_HIGHPRI" or
why WQ is created in hns_roce_v2_init.

If it is WQ_HIGHPRI, yes, it is much better to implement flush operation
ASAP to help generating flushed cqe as ULP may poll cqe urgently. If you
concerned allocation stage, as flush operation is support for hip08 only,
there is no other place proper than here I think.

> 
>> diff --git a/drivers/infiniband/hw/hns/hns_roce_qp.c b/drivers/infiniband/hw/hns/hns_roce_qp.c
>> index bec48f2..2c8f726 100644
>> +++ b/drivers/infiniband/hw/hns/hns_roce_qp.c
>> @@ -43,6 +43,49 @@
>>  
>>  #define SQP_NUM				(2 * HNS_ROCE_MAX_PORTS)
>>  
>> +static void flush_work_handle(struct work_struct *work)
>> +{
>> +	struct hns_roce_flush_work *flush_work = container_of(work,
>> +					struct hns_roce_flush_work, work);
>> +	struct hns_roce_qp *hr_qp = flush_work->hr_qp;
>> +	struct device *dev = flush_work->hr_dev->dev;
>> +	struct ib_qp_attr attr;
>> +	int attr_mask;
>> +	int ret;
>> +
>> +	attr_mask = IB_QP_STATE;
>> +	attr.qp_state = IB_QPS_ERR;
>> +
>> +	ret = hns_roce_modify_qp(&hr_qp->ibqp, &attr, attr_mask, NULL);
>> +	if (ret)
>> +		dev_err(dev, "Modify QP to error state failed(%d) during CQE flush\n",
>> +			ret);
> 
> There is something wrong with your description as all this seems to do
> is tell the HW to go to the ERR state.

For the flush operation, in addition to modify qp to ERR state, the head pointers
of SQ and RQ are also told to the HW with this interface as following. This part
of codes is already there.

if (new_state == IB_QPS_ERR) {
        roce_set_field(context->byte_160_sq_ci_pi,
                       V2_QPC_BYTE_160_SQ_PRODUCER_IDX_M,
                       V2_QPC_BYTE_160_SQ_PRODUCER_IDX_S,
                       hr_qp->sq.head);
        roce_set_field(qpc_mask->byte_160_sq_ci_pi,
                       V2_QPC_BYTE_160_SQ_PRODUCER_IDX_M,
                       V2_QPC_BYTE_160_SQ_PRODUCER_IDX_S, 0);

        if (!ibqp->srq) {
                roce_set_field(context->byte_84_rq_ci_pi,
                       V2_QPC_BYTE_84_RQ_PRODUCER_IDX_M,
                       V2_QPC_BYTE_84_RQ_PRODUCER_IDX_S,
                       hr_qp->rq.head);
                roce_set_field(qpc_mask->byte_84_rq_ci_pi,
                       V2_QPC_BYTE_84_RQ_PRODUCER_IDX_M,
                       V2_QPC_BYTE_84_RQ_PRODUCER_IDX_S, 0);
        }
}

> 
> Why don't you do this from hns_roce_irq_work_handle() ?

As described in the cover letter, here we used CMWQ (concurrency management workqueue)
to make sure that flush operation can be implemented ASAP. Current irq workqueue is
a singlethread workqueue, which may delay the flush too long when the system is heavy.

Do you mean we can change irq workqueue to CMWQ to put flush work into it?

> 
>> +	kfree(flush_work);
>> +
>> +	/*
>> +	 * make sure we signal QP destroy leg that flush QP was completed
>> +	 * so that it can safely proceed ahead now and destroy QP
>> +	 */
>> +	if (atomic_dec_and_test(&hr_qp->refcount))
>> +		complete(&hr_qp->free);
> 
>> +}
>> +
>> +void init_flush_work(struct hns_roce_dev *hr_dev, struct hns_roce_qp *hr_qp)
>> +{
>> +	struct hns_roce_flush_work *flush_work;
>> +
>> +	flush_work = kzalloc(sizeof(struct hns_roce_flush_work), GFP_ATOMIC);
>> +	if (!flush_work)
>> +		return;
> 
> Don't do things that can fail here

Do you mean that as "GFP_ATOMIC" is used, the if branch can be deleted?

> 
>> +
>> +	flush_work->hr_dev = hr_dev;
>> +	flush_work->hr_qp = hr_qp;
>> +	INIT_WORK(&flush_work->work, flush_work_handle);
>> +	atomic_inc(&hr_qp->refcount);
>> +	queue_work(hr_dev->flush_workq, &flush_work->work);
>> +}
>> +
>>  void hns_roce_qp_event(struct hns_roce_dev *hr_dev, u32 qpn, int event_type)
>>  {
>>  	struct device *dev = hr_dev->dev;
> 
> .
>