Re: [PATCH v3 2/5] bus/vdev: add lock on vdev device list

["Burakov, Anatoly" <anatoly.burakov@intel.com>]

On 19-Apr-18 5:50 PM, Jianfeng Tan wrote:
> As we could add virtual devices from different threads now, we
> add a spin lock to protect the vdev device list.
> 
> Suggested-by: Anatoly Burakov <anatoly.burakov@intel.com>
> Signed-off-by: Jianfeng Tan <jianfeng.tan@intel.com>
> Reviewed-by: Qi Zhang <qi.z.zhang@intel.com>
> ---

<...>

> +/* The caller shall be responsible for thread-safe */
>   static struct rte_vdev_device *
>   find_vdev(const char *name)
>   {
> @@ -203,10 +206,6 @@ rte_vdev_init(const char *name, const char *args)
>   	if (name == NULL)
>   		return -EINVAL;
>   
> -	dev = find_vdev(name);
> -	if (dev)
> -		return -EEXIST;
> -
>   	devargs = alloc_devargs(name, args);
>   	if (!devargs)
>   		return -ENOMEM;
> @@ -221,16 +220,28 @@ rte_vdev_init(const char *name, const char *args)
>   	dev->device.numa_node = SOCKET_ID_ANY;
>   	dev->device.name = devargs->name;
>   
> +	rte_spinlock_lock(&vdev_device_list_lock);
> +	if (find_vdev(name)) {
> +		rte_spinlock_unlock(&vdev_device_list_lock);
> +		ret = -EEXIST;
> +		goto fail;
> +	}
> +	TAILQ_INSERT_TAIL(&vdev_device_list, dev, next);
> +	rte_spinlock_unlock(&vdev_device_list_lock);
> +

I wonder if is possible to just leave the tailq locked until you either 
insert the device into tailq, or figure out that it's not possible? 
Seems like doing two locks here is unnecessary, unless 
vdev_probe_all_drivers needs this tailq unlocked...

>   	ret = vdev_probe_all_drivers(dev);
>   	if (ret) {
>   		if (ret > 0)
>   			VDEV_LOG(ERR, "no driver found for %s\n", name);
> +		/* If fails, remove it from vdev list */
> +		rte_spinlock_lock(&vdev_device_list_lock);
> +		TAILQ_REMOVE(&vdev_device_list, dev, next);
> +		rte_spinlock_unlock(&vdev_device_list_lock);
>   		goto fail;
>   	}
>   
>   	TAILQ_INSERT_TAIL(&devargs_list, devargs, next);
>   
> -	TAILQ_INSERT_TAIL(&vdev_device_list, dev, next);
>   	return 0;
>   
>   fail:
> @@ -266,17 +277,25 @@ rte_vdev_uninit(const char *name)
>   	if (name == NULL)
>   		return -EINVAL;
>   
> +	rte_spinlock_lock(&vdev_device_list_lock);
>   	dev = find_vdev(name);
> -	if (!dev)
> +	if (!dev) {
> +		rte_spinlock_unlock(&vdev_device_list_lock);
>   		return -ENOENT;
> +	}
> +	TAILQ_REMOVE(&vdev_device_list, dev, next);
> +	rte_spinlock_unlock(&vdev_device_list_lock);
>   
>   	devargs = dev->device.devargs;
>   
>   	ret = vdev_remove_driver(dev);
> -	if (ret)
> +	if (ret) {
> +		/* If fails, add back to vdev list */
> +		rte_spinlock_lock(&vdev_device_list_lock);
> +		TAILQ_INSERT_TAIL(&vdev_device_list, dev, next);
> +		rte_spinlock_unlock(&vdev_device_list_lock);
>   		return ret;
> -
> -	TAILQ_REMOVE(&vdev_device_list, dev, next);
> +	}

Same comment here - perhaps keep the lock locked all the way? Maybe a 
good way to ensure you don't miss anything is put most of it in a static 
function, and do

static int vdev_uninit() {
	...
}

static int rte_vdev_uninit() {
	int ret;
	lock();
	ret = vdev_uninit();
	unlock();
	return ret;
}

? In general, it is better to do lock/unlock in one place and not 
disperse lock/unlock calls across various branches.

>   
>   	TAILQ_REMOVE(&devargs_list, devargs, next);
>   
> @@ -314,19 +333,25 @@ vdev_scan(void)
>   		if (devargs->bus != &rte_vdev_bus)
>   			continue;
>   
> -		dev = find_vdev(devargs->name);
> -		if (dev)
> -			continue;
> -
>   		dev = calloc(1, sizeof(*dev));
>   		if (!dev)
>   			return -1;
>   
> +		rte_spinlock_lock(&vdev_device_list_lock);
> +
> +		if (find_vdev(devargs->name)) {
> +			rte_spinlock_unlock(&vdev_device_list_lock);
> +			free(dev);
> +			continue;
> +		}
> +
>   		dev->device.devargs = devargs;
>   		dev->device.numa_node = SOCKET_ID_ANY;
>   		dev->device.name = devargs->name;
>   
>   		TAILQ_INSERT_TAIL(&vdev_device_list, dev, next);
> +
> +		rte_spinlock_unlock(&vdev_device_list_lock);
>   	}
>   
>   	return 0;
> @@ -340,6 +365,10 @@ vdev_probe(void)
>   
>   	/* call the init function for each virtual device */
>   	TAILQ_FOREACH(dev, &vdev_device_list, next) {
> +		/* we don't use the vdev lock here, as it's only used in DPDK
> +		 * initialization; and we don't want to hold such a lock when
> +		 * we call each driver probe.
> +		 */
>   
>   		if (dev->device.driver)
>   			continue;
> @@ -360,14 +389,18 @@ vdev_find_device(const struct rte_device *start, rte_dev_cmp_t cmp,
>   {
>   	struct rte_vdev_device *dev;
>   
> +	rte_spinlock_lock(&vdev_device_list_lock);
>   	TAILQ_FOREACH(dev, &vdev_device_list, next) {
>   		if (start && &dev->device == start) {
>   			start = NULL;
>   			continue;
>   		}
> -		if (cmp(&dev->device, data) == 0)
> +		if (cmp(&dev->device, data) == 0) {
> +			rte_spinlock_unlock(&vdev_device_list_lock);
>   			return &dev->device;
> +		}
>   	}
> +	rte_spinlock_unlock(&vdev_device_list_lock);
>   	return NULL;

How about

break;
}
unlock();
return dev ? &dev->device : NULL;

? Seems clearer to me.

>   }
>   
> 


-- 
Thanks,
Anatoly