nbd: fixes for nbd

spinlocked ranges should be small and not contain calls into huge subfunctions. Fix my mistake and just get the pointer to the socket instead of doing everything with spinlock held.

Reported-by Mikulas Patocka <[email protected]>

Signed-off-by Markus Pargmann <[email protected]>

Changelog:
Pranay Kr. Srivastava<[email protected]>:

1) Use spin_lock instead of irq version for sock_shutdown.

2) Use system work queue to actually trigger the shutdown of socket. This solves the issue when kernel_sendmsg is currently blocked while a timeout occurs.

Signed-off-by Pranay Kr. Srivastava <[email protected]>

---
 drivers/block/nbd.c |   62 +++++++++++++++++++++++++++++----------------------
 1 file changed, 35 insertions(+), 27 deletions(-)

drivers/block/nbd.c +1 -0

index 08afbc7..a5dab48 100644
@@ -39,6 +39,7 @@
 #include <asm/types.h>
 
 #include <linux/nbd.h>
+#include <linux/workqueue.h>
 
 struct nbd_device {
 	u32 flags;

drivers/block/nbd.c +23 -26

 static inline struct device *nbd_to_dev(struct nbd_device *nbd)
 {

@@ -172,39 +175,31 @@ static void nbd_end_request(struct nbd_device *nbd, struct request *req)
  */
 static void sock_shutdown(struct nbd_device *nbd)
 {
-	spin_lock_irq(&nbd->sock_lock);
-
-	if (!nbd->sock) {
-		spin_unlock_irq(&nbd->sock_lock);
-		return;
-	}
+	struct socket *sock;
 
-	dev_warn(disk_to_dev(nbd->disk), "shutting down socket\n");
-	kernel_sock_shutdown(nbd->sock, SHUT_RDWR);
-	sockfd_put(nbd->sock);
+	spin_lock(&nbd->sock_lock);
+	sock = nbd->sock;
 	nbd->sock = NULL;
-	spin_unlock_irq(&nbd->sock_lock);
+	spin_unlock(&nbd->sock_lock);
+
+	if (!sock)
+		return;
 
 	del_timer(&nbd->timeout_timer);
+	dev_warn(disk_to_dev(nbd->disk), "shutting down socket\n");
+	kernel_sock_shutdown(sock, SHUT_RDWR);
+	sockfd_put(sock);
 }
 
 static void nbd_xmit_timeout(unsigned long arg)
 {
 	struct nbd_device *nbd = (struct nbd_device *)arg;
-	unsigned long flags;
 
 	if (list_empty(&nbd->queue_head))
 		return;
-
-	spin_lock_irqsave(&nbd->sock_lock, flags);
-
-	nbd->timedout = true;
-
-	if (nbd->sock)
-		kernel_sock_shutdown(nbd->sock, SHUT_RDWR);
-
-	spin_unlock_irqrestore(&nbd->sock_lock, flags);
-
+	atomic_inc(&nbd->timedout);
+	schedule_work(&nbd->ws_nbd);
+	wake_up(&nbd->waiting_wq);
 	dev_err(nbd_to_dev(nbd), "Connection timed out, shutting down connection\n");
 }
 

@@ -592,7 +587,11 @@ static int nbd_thread_send(void *data)
 		spin_unlock_irq(&nbd->queue_lock);
 
 		/* handle request */
-		nbd_handle_req(nbd, req);
+		if (atomic_read(&nbd->timedout)) {
+			req->errors++;
+			nbd_end_request(nbd, req);
+		} else
+			nbd_handle_req(nbd, req);
 	}
 
 	nbd->task_send = NULL;

@@ -666,12 +665,13 @@ out:
 static void nbd_reset(struct nbd_device *nbd)
 {
 	nbd->disconnect = false;
-	nbd->timedout = false;
+	atomic_set(&nbd->timedout, 0);
 	nbd->blksize = 1024;
 	nbd->bytesize = 0;
 	set_capacity(nbd->disk, 0);
 	nbd->flags = 0;
 	nbd->xmit_timeout = 0;
+	INIT_WORK(&nbd->ws_nbd, nbd_work_func);
 	queue_flag_clear_unlocked(QUEUE_FLAG_DISCARD, nbd->disk->queue);
 	del_timer_sync(&nbd->timeout_timer);
 }

@@ -804,16 +804,16 @@ static int __nbd_ioctl(struct block_device *bdev, struct nbd_device *nbd,
 		nbd_dev_dbg_close(nbd);
 		kthread_stop(thread);
 
-		mutex_lock(&nbd->tx_lock);
-
 		sock_shutdown(nbd);
+		mutex_lock(&nbd->tx_lock);
 		nbd_clear_que(nbd);
 		kill_bdev(bdev);
 		nbd_bdev_reset(bdev);
 
 		if (nbd->disconnect) /* user requested, ignore socket errors */
 			error = 0;
-		if (nbd->timedout)
+
+		if (atomic_read(&nbd->timedout))
 			error = -ETIMEDOUT;
 
 		nbd_reset(nbd);

When a timeout occurs or a recv fails, then instead of abruplty killing nbd block device wait for it's users to finish.

This is more required when filesystem(s) like ext2 or ext3 don't expect their buffer heads to disappear while the filesystem is mounted.

Use a kref for users using this. The device will be released for kref count of 2, not less or more.

Signed-off-by Pranay Kr. Srivastava <[email protected]>

---
 drivers/block/nbd.c |   51 +++++++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 51 insertions(+)

drivers/block/nbd.c +30 -0

index af86c9b..59db890 100644
@@ -71,6 +71,8 @@ struct nbd_device {
 	struct dentry *dbg_dir;
 #endif
 	struct work_struct ws_nbd;
+	struct kref users;
+	struct completion user_completion;
 };
 
 #if IS_ENABLED(CONFIG_DEBUG_FS)

@@ -674,6 +676,7 @@ static void nbd_reset(struct nbd_device *nbd)
 	nbd->flags = 0;
 	nbd->xmit_timeout = 0;
 	INIT_WORK(&nbd->ws_nbd, nbd_work_func);
+	init_completion(&nbd->user_completion);
 	queue_flag_clear_unlocked(QUEUE_FLAG_DISCARD, nbd->disk->queue);
 	del_timer_sync(&nbd->timeout_timer);
 }

@@ -807,6 +810,7 @@ static int __nbd_ioctl(struct block_device *bdev, struct nbd_device *nbd,
 		kthread_stop(thread);
 
 		sock_shutdown(nbd);
+		wait_for_completion(&nbd->user_completion);
 		mutex_lock(&nbd->tx_lock);
 		nbd_clear_que(nbd);
 		kill_bdev(bdev);

@@ -858,12 +862,58 @@ static int nbd_ioctl(struct block_device *bdev, fmode_t mode,
 	return error;
 }
 
+static void nbd_kref_release(struct kref *kref_users)
+{
+	struct nbd_device *nbd = container_of(kref_users, struct nbd_device,
+			users);
+	pr_debug("Releasing kref [%s]\n", __FUNCTION__);
+	complete(&nbd->user_completion);
+
+}
+
+static int nbd_open(struct block_device *bdev, fmode_t mode)
+{
+	struct nbd_device *nbd_dev = bdev->bd_disk->private_data;
+
+	kref_get(&nbd_dev->users);
+	pr_debug("Opening nbd_dev %s. Active users = %u\n",
+			bdev->bd_disk->disk_name,
+			atomic_read(&nbd_dev->users.refcount) - 1);
+	return 0;
+}
+
+static void nbd_release(struct gendisk *disk, fmode_t mode)
+{
+	struct nbd_device *nbd_dev = disk->private_data;
+	/*
+	*kref_init initializes ref count to 1, so we
+	*we check for refcount to be 2 for a final put.

drivers/block/nbd.c +11 -0

+	}
+
+	pr_debug("Closing nbd_dev %s. Active users = %u\n",
+			disk->disk_name,
+			atomic_read(&nbd_dev->users.refcount) - 1);
+}
+
 static const struct block_device_operations nbd_fops = {
 	.owner =	THIS_MODULE,
 	.ioctl =	nbd_ioctl,
 	.compat_ioctl =	nbd_ioctl,
+	.open = 	nbd_open,
+	.release = 	nbd_release
 };
 
+
 static void nbd_work_func(struct work_struct *ws_nbd)
 {
 	struct nbd_device *nbd_dev = container_of(ws_nbd, struct nbd_device,

@@ -1098,6 +1148,7 @@ static int __init nbd_init(void)
 		disk->first_minor = i << part_shift;
 		disk->fops = &nbd_fops;
 		disk->private_data = &nbd_dev[i];
+		kref_init(&nbd_dev[i].users);
 		sprintf(disk->disk_name, "nbd%d", i);
 		nbd_reset(&nbd_dev[i]);
 		add_disk(disk);

This patch series fixes the following

1) fix might_sleep warning on socket shutdown:
Fix sock_shutdown to avoid calling kernel_sock_shutdown while holding spin_lock.

2)fix various coding standard warnings
Make shutdown get called in a process context instead, using system_wq.

3) make nbd device wait for its users.
When a timeout or error occurs then nbd driver simply kills the block device. Many filesystem(s) example ext2/ext3 don't expect their buffer heads to disappear like that. Fix this by making nbd device wait for its users.

4) use device_attr macros for sysfs attribute
use DEVICE_ATTR_RO for sysfs pid attribute.

Pranay Kr. Srivastava (4):
fix might_sleep warning on socket shutdown.
fix various coding standard warnings
make nbd device wait for its users.
use device_attr macros for sysfs attribute

 drivers/block/nbd.c |  150 +++++++++++++++++++++++++++++++++++----------------
 1 file changed, 105 insertions(+), 45 deletions(-)

--
1.7.9.5

This patch series fixes the following

1) fix might_sleep warning on socket shutdown:
Fix sock_shutdown to avoid calling kernel_sock_shutdown while holding spin_lock.

2)fix various coding standard warnings
Make shutdown get called in a process context instead, using system_wq.

3) make nbd device wait for its users.
When a timeout or error occurs then nbd driver simply kills the block device. Many filesystem(s) example ext2/ext3 don't expect their buffer heads to disappear like that. Fix this by making nbd device wait for its users.

4) use device_attr macros for sysfs attribute
use DEVICE_ATTR_RO for sysfs pid attribute.

spinlocked ranges should be small and not contain calls into huge subfunctions. Fix my mistake and just get the pointer to the socket instead of doing everything with spinlock held.

Reported-by Mikulas Patocka <[email protected]>

Signed-off-by Markus Pargmann <[email protected]>

Changelog:
Pranay Kr. Srivastava<[email protected]>:

1) Use spin_lock instead of irq version for sock_shutdown.

2) Use system work queue to actually trigger the shutdown of socket. This solves the issue when kernel_sendmsg is currently blocked while a timeout occurs.

Signed-off-by Pranay Kr. Srivastava <[email protected]>

---
 drivers/block/nbd.c | 65 ++++++++++++++++++++++++++++++++++-------------------
 1 file changed, 42 insertions(+), 23 deletions(-)

drivers/block/nbd.c +42 -24

index 31e73a7..0339d40 100644
@@ -39,6 +39,7 @@
 #include <asm/types.h>
 
 #include <linux/nbd.h>
+#include <linux/workqueue.h>
 
 struct nbd_device {
        u32 flags;

@@ -69,6 +70,10 @@ struct nbd_device {
 #if IS_ENABLED(CONFIG_DEBUG_FS)
        struct dentry *dbg_dir;
 #endif
+       /*
+        *This is specifically for calling sock_shutdown, for now.
+        */
+       struct work_struct ws_shutdown;
 };
 
 #if IS_ENABLED(CONFIG_DEBUG_FS)

@@ -95,6 +100,11 @@ static int max_part;
  */
 static DEFINE_SPINLOCK(nbd_lock);
 
+/*
+ * Shutdown function for nbd_dev work struct.
+ */
+static void nbd_ws_func_shutdown(struct work_struct *);
+
 static inline struct device *nbd_to_dev(struct nbd_device *nbd)
 {
        return disk_to_dev(nbd->disk);

@@ -172,39 +182,35 @@ static void nbd_end_request(struct nbd_device *nbd, struct request *req)
  */
 static void sock_shutdown(struct nbd_device *nbd)
 {
-       spin_lock_irq(&nbd->sock_lock);
-
-       if (!nbd->sock) {
-               spin_unlock_irq(&nbd->sock_lock);
-               return;
-       }
+       struct socket *sock;
 
-       dev_warn(disk_to_dev(nbd->disk), "shutting down socket\n");
-       kernel_sock_shutdown(nbd->sock, SHUT_RDWR);
-       sockfd_put(nbd->sock);
+       spin_lock(&nbd->sock_lock);
+       sock = nbd->sock;
        nbd->sock = NULL;
-       spin_unlock_irq(&nbd->sock_lock);
+       spin_unlock(&nbd->sock_lock);
+
+       if (!sock)
+               return;
 
        del_timer(&nbd->timeout_timer);
+       dev_warn(disk_to_dev(nbd->disk), "shutting down socket\n");
+       kernel_sock_shutdown(sock, SHUT_RDWR);
+       sockfd_put(sock);
 }
 
 static void nbd_xmit_timeout(unsigned long arg)
 {
        struct nbd_device *nbd = (struct nbd_device *)arg;
-       unsigned long flags;
 
        if (list_empty(&nbd->queue_head))
                return;
-
-       spin_lock_irqsave(&nbd->sock_lock, flags);
-
        nbd->timedout = true;
-
-       if (nbd->sock)
-               kernel_sock_shutdown(nbd->sock, SHUT_RDWR);
-
-       spin_unlock_irqrestore(&nbd->sock_lock, flags);
-
+       schedule_work(&nbd->ws_shutdown);
+       /*
+        * Make sure sender thread sees nbd->timedout.
+        */
+       smp_wmb();
+       wake_up(&nbd->waiting_wq);
        dev_err(nbd_to_dev(nbd), "Connection timed out, shutting down connection\n");
 }
 

@@ -592,7 +598,11 @@ static int nbd_thread_send(void *data)
                spin_unlock_irq(&nbd->queue_lock);
 
                /* handle request */
-               nbd_handle_req(nbd, req);
+               if (nbd->timedout) {
+                       req->errors++;
+                       nbd_end_request(nbd, req);
+               } else
+                       nbd_handle_req(nbd, req);
        }
 
        nbd->task_send = NULL;

@@ -672,6 +682,7 @@ static void nbd_reset(struct nbd_device *nbd)
        set_capacity(nbd->disk, 0);
        nbd->flags = 0;
        nbd->xmit_timeout = 0;
+       INIT_WORK(&nbd->ws_shutdown, nbd_ws_func_shutdown);
        queue_flag_clear_unlocked(QUEUE_FLAG_DISCARD, nbd->disk->queue);
        del_timer_sync(&nbd->timeout_timer);
 }

@@ -804,15 +815,15 @@ static int __nbd_ioctl(struct block_device *bdev, struct nbd_device *nbd,
                nbd_dev_dbg_close(nbd);
                kthread_stop(thread);
 
-               mutex_lock(&nbd->tx_lock);
-
                sock_shutdown(nbd);
+               mutex_lock(&nbd->tx_lock);
                nbd_clear_que(nbd);
                kill_bdev(bdev);
                nbd_bdev_reset(bdev);
 
                if (nbd->disconnect) /* user requested, ignore socket errors */
                        error = 0;
+
                if (nbd->timedout)
                        error = -ETIMEDOUT;
 

@@ -863,6 +874,14 @@ static const struct block_device_operations nbd_fops =
        .compat_ioctl = nbd_ioctl,
 };
 
+static void nbd_ws_func_shutdown(struct work_struct *ws_nbd)
+{
+       struct nbd_device *nbd_dev = container_of(ws_nbd, struct nbd_device,
+                                                       ws_shutdown);
+
+       sock_shutdown(nbd_dev);
+}
+
 #if IS_ENABLED(CONFIG_DEBUG_FS)
 
 static int nbd_dbg_tasks_show(struct seq_file *s, void *unused)
--

2.6.2

Hello

Quoted 167 lines · Pranay Kr. Srivastava

spinlocked ranges should be small and not contain calls into huge subfunctions. Fix my mistake and just get the pointer to the socket instead of doing everything with spinlock held.

Reported-by Mikulas Patocka <[email protected]>

Signed-off-by Markus Pargmann <[email protected]>

Changelog:
Pranay Kr. Srivastava<[email protected]>:

1) Use spin_lock instead of irq version for sock_shutdown.

2) Use system work queue to actually trigger the shutdown of socket. This solves the issue when kernel_sendmsg is currently blocked while a timeout occurs.

Signed-off-by Pranay Kr. Srivastava <[email protected]>

---
 drivers/block/nbd.c | 65 ++++++++++++++++++++++++++++++++++-------------------
 1 file changed, 42 insertions(+), 23 deletions(-)

drivers/block/nbd.c +42 -24

index 31e73a7..0339d40 100644
@@ -39,6 +39,7 @@
 #include <asm/types.h>
 
 #include <linux/nbd.h>
+#include <linux/workqueue.h>
 
 struct nbd_device {
        u32 flags;

@@ -69,6 +70,10 @@ struct nbd_device {
 #if IS_ENABLED(CONFIG_DEBUG_FS)
        struct dentry *dbg_dir;
 #endif
+       /*
+        *This is specifically for calling sock_shutdown, for now.
+        */
+       struct work_struct ws_shutdown;
 };
 
 #if IS_ENABLED(CONFIG_DEBUG_FS)

@@ -95,6 +100,11 @@ static int max_part;
  */
 static DEFINE_SPINLOCK(nbd_lock);
 
+/*
+ * Shutdown function for nbd_dev work struct.
+ */
+static void nbd_ws_func_shutdown(struct work_struct *);
+
 static inline struct device *nbd_to_dev(struct nbd_device *nbd)
 {
        return disk_to_dev(nbd->disk);

@@ -172,39 +182,35 @@ static void nbd_end_request(struct nbd_device *nbd, struct request *req)
  */
 static void sock_shutdown(struct nbd_device *nbd)
 {
-       spin_lock_irq(&nbd->sock_lock);
-
-       if (!nbd->sock) {
-               spin_unlock_irq(&nbd->sock_lock);
-               return;
-       }
+       struct socket *sock;
 
-       dev_warn(disk_to_dev(nbd->disk), "shutting down socket\n");
-       kernel_sock_shutdown(nbd->sock, SHUT_RDWR);
-       sockfd_put(nbd->sock);
+       spin_lock(&nbd->sock_lock);
+       sock = nbd->sock;
        nbd->sock = NULL;
-       spin_unlock_irq(&nbd->sock_lock);
+       spin_unlock(&nbd->sock_lock);
+
+       if (!sock)
+               return;
 
        del_timer(&nbd->timeout_timer);
+       dev_warn(disk_to_dev(nbd->disk), "shutting down socket\n");
+       kernel_sock_shutdown(sock, SHUT_RDWR);
+       sockfd_put(sock);
 }
 
 static void nbd_xmit_timeout(unsigned long arg)
 {
        struct nbd_device *nbd = (struct nbd_device *)arg;
-       unsigned long flags;
 
        if (list_empty(&nbd->queue_head))
                return;
-
-       spin_lock_irqsave(&nbd->sock_lock, flags);
-
        nbd->timedout = true;
-
-       if (nbd->sock)
-               kernel_sock_shutdown(nbd->sock, SHUT_RDWR);
-
-       spin_unlock_irqrestore(&nbd->sock_lock, flags);
-
+       schedule_work(&nbd->ws_shutdown);
+       /*
+        * Make sure sender thread sees nbd->timedout.
+        */
+       smp_wmb();
+       wake_up(&nbd->waiting_wq);
        dev_err(nbd_to_dev(nbd), "Connection timed out, shutting down connection\n");
 }
 

@@ -592,7 +598,11 @@ static int nbd_thread_send(void *data)
                spin_unlock_irq(&nbd->queue_lock);
 
                /* handle request */
-               nbd_handle_req(nbd, req);
+               if (nbd->timedout) {
+                       req->errors++;
+                       nbd_end_request(nbd, req);
+               } else
+                       nbd_handle_req(nbd, req);
        }
 
        nbd->task_send = NULL;

@@ -672,6 +682,7 @@ static void nbd_reset(struct nbd_device *nbd)
        set_capacity(nbd->disk, 0);
        nbd->flags = 0;
        nbd->xmit_timeout = 0;
+       INIT_WORK(&nbd->ws_shutdown, nbd_ws_func_shutdown);
        queue_flag_clear_unlocked(QUEUE_FLAG_DISCARD, nbd->disk->queue);
        del_timer_sync(&nbd->timeout_timer);
 }

@@ -804,15 +815,15 @@ static int __nbd_ioctl(struct block_device *bdev, struct nbd_device *nbd,
                nbd_dev_dbg_close(nbd);
                kthread_stop(thread);
 
-               mutex_lock(&nbd->tx_lock);
-
                sock_shutdown(nbd);
+               mutex_lock(&nbd->tx_lock);
                nbd_clear_que(nbd);
                kill_bdev(bdev);
                nbd_bdev_reset(bdev);
 
                if (nbd->disconnect) /* user requested, ignore socket errors */
                        error = 0;
+
                if (nbd->timedout)
                        error = -ETIMEDOUT;
 

@@ -863,6 +874,14 @@ static const struct block_device_operations nbd_fops =
        .compat_ioctl = nbd_ioctl,
 };
 
+static void nbd_ws_func_shutdown(struct work_struct *ws_nbd)
+{
+       struct nbd_device *nbd_dev = container_of(ws_nbd, struct nbd_device,
+                                                       ws_shutdown);
+
+       sock_shutdown(nbd_dev);
+}
+
 #if IS_ENABLED(CONFIG_DEBUG_FS)
 
 static int nbd_dbg_tasks_show(struct seq_file *s, void *unused)
--

Quoted 2 lines

2.6.2

Any update for the above patch series?
--

        ---P.K.S

spinlocked ranges should be small and not contain calls into huge subfunctions. Fix my mistake and just get the pointer to the socket instead of doing everything with spinlock held.

Reported-by Mikulas Patocka <[email protected]>

Signed-off-by Markus Pargmann <[email protected]>

Changelog:
Pranay Kr. Srivastava<[email protected]>:

1) Use spin_lock instead of irq version for sock_shutdown.

2) Use system work queue to actually trigger the shutdown of socket. This solves the issue when kernel_sendmsg is currently blocked while a timeout occurs.

Signed-off-by Pranay Kr. Srivastava <[email protected]>

---
 drivers/block/nbd.c | 65 ++++++++++++++++++++++++++++++++++-------------------
 1 file changed, 42 insertions(+), 23 deletions(-)

drivers/block/nbd.c +20 -15

index 31e73a7..0339d40 100644
@@ -39,6 +39,7 @@
 #include <asm/types.h>
 
 #include <linux/nbd.h>
+#include <linux/workqueue.h>
 
 struct nbd_device {
 	u32 flags;

@@ -69,6 +70,10 @@ struct nbd_device {
 #if IS_ENABLED(CONFIG_DEBUG_FS)
 	struct dentry *dbg_dir;
 #endif
+	/*
+	 *This is specifically for calling sock_shutdown, for now.
+	 */
+	struct work_struct ws_shutdown;
 };
 
 #if IS_ENABLED(CONFIG_DEBUG_FS)

@@ -95,6 +100,11 @@ static int max_part;
  */
 static DEFINE_SPINLOCK(nbd_lock);
 
+/*
+ * Shutdown function for nbd_dev work struct.
+ */
+static void nbd_ws_func_shutdown(struct work_struct *);
+
 static inline struct device *nbd_to_dev(struct nbd_device *nbd)
 {
 	return disk_to_dev(nbd->disk);

@@ -172,39 +182,35 @@ static void nbd_end_request(struct nbd_device *nbd, struct request *req)
  */
 static void sock_shutdown(struct nbd_device *nbd)
 {
-	spin_lock_irq(&nbd->sock_lock);
-
-	if (!nbd->sock) {
-		spin_unlock_irq(&nbd->sock_lock);
-		return;
-	}
+	struct socket *sock;
 
-	dev_warn(disk_to_dev(nbd->disk), "shutting down socket\n");
-	kernel_sock_shutdown(nbd->sock, SHUT_RDWR);
-	sockfd_put(nbd->sock);
+	spin_lock(&nbd->sock_lock);
+	sock = nbd->sock;
 	nbd->sock = NULL;
-	spin_unlock_irq(&nbd->sock_lock);
+	spin_unlock(&nbd->sock_lock);
+
+	if (!sock)
+		return;
 
 	del_timer(&nbd->timeout_timer);
+	dev_warn(disk_to_dev(nbd->disk), "shutting down socket\n");
+	kernel_sock_shutdown(sock, SHUT_RDWR);
+	sockfd_put(sock);
 }
 
 static void nbd_xmit_timeout(unsigned long arg)
 {
 	struct nbd_device *nbd = (struct nbd_device *)arg;
-	unsigned long flags;
 
 	if (list_empty(&nbd->queue_head))
 		return;
-
-	spin_lock_irqsave(&nbd->sock_lock, flags);
-
 	nbd->timedout = true;
-

drivers/block/nbd.c +7 -2

+	wake_up(&nbd->waiting_wq);
 	dev_err(nbd_to_dev(nbd), "Connection timed out, shutting down connection\n");
 }
 

@@ -592,7 +598,11 @@ static int nbd_thread_send(void *data)
 		spin_unlock_irq(&nbd->queue_lock);
 
 		/* handle request */
-		nbd_handle_req(nbd, req);
+		if (nbd->timedout) {
+			req->errors++;
+			nbd_end_request(nbd, req);
+		} else
+			nbd_handle_req(nbd, req);
 	}
 
 	nbd->task_send = NULL;

@@ -672,6 +682,7 @@ static void nbd_reset(struct nbd_device *nbd)
 	set_capacity(nbd->disk, 0);
 	nbd->flags = 0;
 	nbd->xmit_timeout = 0;
+	INIT_WORK(&nbd->ws_shutdown, nbd_ws_func_shutdown);
 	queue_flag_clear_unlocked(QUEUE_FLAG_DISCARD, nbd->disk->queue);
 	del_timer_sync(&nbd->timeout_timer);
 }

@@ -804,15 +815,15 @@ static int __nbd_ioctl(struct block_device *bdev, struct nbd_device *nbd,
 		nbd_dev_dbg_close(nbd);
 		kthread_stop(thread);
 
-		mutex_lock(&nbd->tx_lock);

drivers/block/nbd.c +8 -0

 		if (nbd->timedout)
 			error = -ETIMEDOUT;
 

@@ -863,6 +874,14 @@ static const struct block_device_operations nbd_fops =
 	.compat_ioctl =	nbd_ioctl,
 };
 
+static void nbd_ws_func_shutdown(struct work_struct *ws_nbd)
+{
+	struct nbd_device *nbd_dev = container_of(ws_nbd, struct nbd_device,
+							ws_shutdown);
+
+	sock_shutdown(nbd_dev);
+}
+
 #if IS_ENABLED(CONFIG_DEBUG_FS)
 
 static int nbd_dbg_tasks_show(struct seq_file *s, void *unused)

Hi,

Quoted 16 lines · On Thursday 02 June 2016 13:24:57 Pranay Kr. Srivastava wrote:

spinlocked ranges should be small and not contain calls into huge subfunctions. Fix my mistake and just get the pointer to the socket instead of doing everything with spinlock held.

Reported-by Mikulas Patocka <[email protected]>

Signed-off-by Markus Pargmann <[email protected]>

Changelog:
Pranay Kr. Srivastava<[email protected]>:

1) Use spin_lock instead of irq version for sock_shutdown.

2) Use system work queue to actually trigger the shutdown of socket. This solves the issue when kernel_sendmsg is currently blocked while a timeout occurs.

Quoted 1 line

Signed-off-by Pranay Kr. Srivastava <[email protected]>

This looks better. Some smaller things inline. Also this patch does not apply on my tree anymore. Can you please rebase it onto:
http://git.pengutronix.de/?p=mpa/linux-nbd.git;a=shortlog;h=refs/heads/master

Quoted 81 lines

---
 drivers/block/nbd.c | 65 ++++++++++++++++++++++++++++++++++-------------------
 1 file changed, 42 insertions(+), 23 deletions(-)

drivers/block/nbd.c +20 -15

index 31e73a7..0339d40 100644
@@ -39,6 +39,7 @@
 #include <asm/types.h>
 
 #include <linux/nbd.h>
+#include <linux/workqueue.h>
 
 struct nbd_device {
      u32 flags;

@@ -69,6 +70,10 @@ struct nbd_device {
 #if IS_ENABLED(CONFIG_DEBUG_FS)
      struct dentry *dbg_dir;
 #endif
+     /*
+      *This is specifically for calling sock_shutdown, for now.
+      */
+     struct work_struct ws_shutdown;
 };
 
 #if IS_ENABLED(CONFIG_DEBUG_FS)

@@ -95,6 +100,11 @@ static int max_part;
  */
 static DEFINE_SPINLOCK(nbd_lock);
 
+/*
+ * Shutdown function for nbd_dev work struct.
+ */
+static void nbd_ws_func_shutdown(struct work_struct *);
+
 static inline struct device *nbd_to_dev(struct nbd_device *nbd)
 {
      return disk_to_dev(nbd->disk);

@@ -172,39 +182,35 @@ static void nbd_end_request(struct nbd_device *nbd, struct request *req)
  */
 static void sock_shutdown(struct nbd_device *nbd)
 {
-     spin_lock_irq(&nbd->sock_lock);
-
-     if (!nbd->sock) {
-             spin_unlock_irq(&nbd->sock_lock);
-             return;
-     }
+     struct socket *sock;
 
-     dev_warn(disk_to_dev(nbd->disk), "shutting down socket\n");
-     kernel_sock_shutdown(nbd->sock, SHUT_RDWR);
-     sockfd_put(nbd->sock);
+     spin_lock(&nbd->sock_lock);
+     sock = nbd->sock;
      nbd->sock = NULL;
-     spin_unlock_irq(&nbd->sock_lock);
+     spin_unlock(&nbd->sock_lock);
+
+     if (!sock)
+             return;
 
      del_timer(&nbd->timeout_timer);
+     dev_warn(disk_to_dev(nbd->disk), "shutting down socket\n");
+     kernel_sock_shutdown(sock, SHUT_RDWR);
+     sockfd_put(sock);
 }
 
 static void nbd_xmit_timeout(unsigned long arg)
 {
      struct nbd_device *nbd = (struct nbd_device *)arg;
-     unsigned long flags;
 
      if (list_empty(&nbd->queue_head))
              return;
-
-     spin_lock_irqsave(&nbd->sock_lock, flags);
-
      nbd->timedout = true;
-

drivers/block/nbd.c static void nbd_end_request(struct nbd_device *nbd, struct request *req) · line 212

Quoted 10 lines

drivers/block/nbd.c +5 -5

-     if (nbd->sock)
-             kernel_sock_shutdown(nbd->sock, SHUT_RDWR);
-
-     spin_unlock_irqrestore(&nbd->sock_lock, flags);
-
+     schedule_work(&nbd->ws_shutdown);
+     /*
+      * Make sure sender thread sees nbd->timedout.
+      */
+     smp_wmb();

I am not sure that we need this memory barrier here. But as it is just the timeout path it probably won't hurt.

Quoted 37 lines

drivers/block/nbd.c +8 -3

+     wake_up(&nbd->waiting_wq);
      dev_err(nbd_to_dev(nbd), "Connection timed out, shutting down connection\n");
 }
 

@@ -592,7 +598,11 @@ static int nbd_thread_send(void *data)
              spin_unlock_irq(&nbd->queue_lock);
 
              /* handle request */
-             nbd_handle_req(nbd, req);
+             if (nbd->timedout) {
+                     req->errors++;
+                     nbd_end_request(nbd, req);
+             } else
+                     nbd_handle_req(nbd, req);
      }
 
      nbd->task_send = NULL;

@@ -672,6 +682,7 @@ static void nbd_reset(struct nbd_device *nbd)
      set_capacity(nbd->disk, 0);
      nbd->flags = 0;
      nbd->xmit_timeout = 0;
+     INIT_WORK(&nbd->ws_shutdown, nbd_ws_func_shutdown);
      queue_flag_clear_unlocked(QUEUE_FLAG_DISCARD, nbd->disk->queue);
      del_timer_sync(&nbd->timeout_timer);
 }

@@ -804,15 +815,15 @@ static int __nbd_ioctl(struct block_device *bdev, struct nbd_device *nbd,
              nbd_dev_dbg_close(nbd);
              kthread_stop(thread);
 
-             mutex_lock(&nbd->tx_lock);
-
              sock_shutdown(nbd);
+             mutex_lock(&nbd->tx_lock);
              nbd_clear_que(nbd);
              kill_bdev(bdev);
              nbd_bdev_reset(bdev);
 

drivers/block/nbd.c static int __nbd_ioctl(struct block_device *bdev, struct nbd_device *nbd, · line 826

Quoted 3 lines

drivers/block/nbd.c +1 -0

              if (nbd->disconnect) /* user requested, ignore socket errors */
                      error = 0;
+

Random newline here.

Best Regards,

Markus

Quoted 9 lines

drivers/block/nbd.c +1 -0

              if (nbd->timedout)
                      error = -ETIMEDOUT;
 

@@ -863,6 +874,14 @@ static const struct block_device_operations nbd_fops =
      .compat_ioctl = nbd_ioctl,
 };
 
+static void nbd_ws_func_shutdown(struct work_struct *ws_nbd)

drivers/block/nbd.c static const struct block_device_operations nbd_fops = · line 887

Quoted 10 lines

drivers/block/nbd.c +7 -0

+{
+     struct nbd_device *nbd_dev = container_of(ws_nbd, struct nbd_device,
+                                                     ws_shutdown);
+
+     sock_shutdown(nbd_dev);
+}
+
 #if IS_ENABLED(CONFIG_DEBUG_FS)
 
 static int nbd_dbg_tasks_show(struct seq_file *s, void *unused)

--

Pengutronix e.K.                           |                             |
Industrial Linux Solutions                 | http://www.pengutronix.de/  |
Peiner Str. 6-8, 31137 Hildesheim, Germany | Phone: +49-5121-206917-0    |
Amtsgericht Hildesheim, HRA 2686           | Fax:   +49-5121-206917-5555 |

spinlocked ranges should be small and not contain calls into huge subfunctions. Fix my mistake and just get the pointer to the socket instead of doing everything with spinlock held.

Reported-by Mikulas Patocka <[email protected]>

Signed-off-by Markus Pargmann <[email protected]>

Changelog:
Pranay Kr. Srivastava<[email protected]>:

1) Use spin_lock instead of irq version for sock_shutdown.

2) Use system work queue to actually trigger the shutdown of socket. This solves the issue when kernel_sendmsg is currently blocked while a timeout occurs.

Signed-off-by Pranay Kr. Srivastava <[email protected]>

---
 drivers/block/nbd.c | 69 ++++++++++++++++++++++++++++++++++-------------------
 1 file changed, 44 insertions(+), 25 deletions(-)

drivers/block/nbd.c +44 -26

index 56f7f5d..586d946 100644
@@ -39,6 +39,7 @@
 #include <asm/types.h>
 
 #include <linux/nbd.h>
+#include <linux/workqueue.h>
 
 struct nbd_device {
        u32 flags;

@@ -69,6 +70,10 @@ struct nbd_device {
 #if IS_ENABLED(CONFIG_DEBUG_FS)
        struct dentry *dbg_dir;
 #endif
+       /*
+       *This is specifically for calling sock_shutdown, for now.
+       */
+       struct work_struct ws_shutdown;
 };
 
 #if IS_ENABLED(CONFIG_DEBUG_FS)

@@ -95,6 +100,11 @@ static int max_part;
  */
 static DEFINE_SPINLOCK(nbd_lock);
 
+/*
+ * Shutdown function for nbd_dev work struct.
+ */
+static void nbd_ws_func_shutdown(struct work_struct *);
+
 static inline struct device *nbd_to_dev(struct nbd_device *nbd)
 {
        return disk_to_dev(nbd->disk);

@@ -172,39 +182,35 @@ static void nbd_end_request(struct nbd_device *nbd, struct request *req)
  */
 static void sock_shutdown(struct nbd_device *nbd)
 {
-       spin_lock_irq(&nbd->sock_lock);
-
-       if (!nbd->sock) {
-               spin_unlock_irq(&nbd->sock_lock);
-               return;
-       }
+       struct socket *sock;
 
-       dev_warn(disk_to_dev(nbd->disk), "shutting down socket\n");
-       kernel_sock_shutdown(nbd->sock, SHUT_RDWR);
-       sockfd_put(nbd->sock);
+       spin_lock(&nbd->sock_lock);
+       sock = nbd->sock;
        nbd->sock = NULL;
-       spin_unlock_irq(&nbd->sock_lock);
+       spin_unlock(&nbd->sock_lock);
+
+       if (!sock)
+               return;
 
        del_timer(&nbd->timeout_timer);
+       dev_warn(disk_to_dev(nbd->disk), "shutting down socket\n");
+       kernel_sock_shutdown(sock, SHUT_RDWR);
+       sockfd_put(sock);
 }
 
 static void nbd_xmit_timeout(unsigned long arg)
 {
        struct nbd_device *nbd = (struct nbd_device *)arg;
-       unsigned long flags;
 
        if (list_empty(&nbd->queue_head))
                return;
-
-       spin_lock_irqsave(&nbd->sock_lock, flags);
-
        nbd->timedout = true;
-
-       if (nbd->sock)
-               kernel_sock_shutdown(nbd->sock, SHUT_RDWR);
-
-       spin_unlock_irqrestore(&nbd->sock_lock, flags);
-
+       schedule_work(&nbd->ws_shutdown);
+       /*
+        * Make sure sender thread sees nbd->timedout.
+        */
+       smp_wmb();
+       wake_up(&nbd->waiting_wq);
        dev_err(nbd_to_dev(nbd), "Connection timed out, shutting down connection\n");
 }
 

@@ -574,8 +580,8 @@ static int nbd_thread_send(void *data)
        while (!kthread_should_stop() || !list_empty(&nbd->waiting_queue)) {
                /* wait for something to do */
                wait_event_interruptible(nbd->waiting_wq,
-                                        kthread_should_stop() ||
-                                        !list_empty(&nbd->waiting_queue));
+                               kthread_should_stop() ||
+                               !list_empty(&nbd->waiting_queue));
 
                /* extract request */
                if (list_empty(&nbd->waiting_queue))

@@ -583,12 +589,16 @@ static int nbd_thread_send(void *data)
 
                spin_lock_irq(&nbd->queue_lock);
                req = list_entry(nbd->waiting_queue.next, struct request,
-                                queuelist);
+                               queuelist);
                list_del_init(&req->queuelist);
                spin_unlock_irq(&nbd->queue_lock);
 
-               /* handle request */
                nbd_handle_req(nbd, req);
+               if (nbd->timedout) {
+                       req->errors++;
+                       nbd_end_request(nbd, req);
+               } else
+                       nbd_handle_req(nbd, req);
        }
 
        nbd->task_send = NULL;

@@ -668,6 +678,7 @@ static void nbd_reset(struct nbd_device *nbd)
        set_capacity(nbd->disk, 0);
        nbd->flags = 0;
        nbd->xmit_timeout = 0;
+       INIT_WORK(&nbd->ws_shutdown, nbd_ws_func_shutdown);
        queue_flag_clear_unlocked(QUEUE_FLAG_DISCARD, nbd->disk->queue);
        del_timer_sync(&nbd->timeout_timer);
 }

@@ -802,11 +813,11 @@ static int __nbd_ioctl(struct block_device *bdev, struct nbd_device *nbd,
                error = nbd_thread_recv(nbd, bdev);
                nbd_dev_dbg_close(nbd);
                kthread_stop(thread);
+               sock_shutdown(nbd);
 
                mutex_lock(&nbd->tx_lock);
                nbd->task_recv = NULL;
 
-               sock_shutdown(nbd);
                nbd_clear_que(nbd);
                kill_bdev(bdev);
                nbd_bdev_reset(bdev);

@@ -862,6 +873,14 @@ static const struct block_device_operations nbd_fops = {
        .compat_ioctl = nbd_ioctl,
 };
 
+static void nbd_ws_func_shutdown(struct work_struct *ws_nbd)
+{
+       struct nbd_device *nbd_dev = container_of(ws_nbd, struct nbd_device,
+                       ws_shutdown);
+
+       sock_shutdown(nbd_dev);
+}
+
 #if IS_ENABLED(CONFIG_DEBUG_FS)
 
 static int nbd_dbg_tasks_show(struct seq_file *s, void *unused)
--

1.9.1

spinlocked ranges should be small and not contain calls into huge subfunctions. Fix my mistake and just get the pointer to the socket instead of doing everything with spinlock held.

Reported-by Mikulas Patocka <[email protected]>

Signed-off-by Markus Pargmann <[email protected]>

Changelog:
Pranay Kr. Srivastava<[email protected]>:

1) Use spin_lock instead of irq version for sock_shutdown.

2) Use system work queue to actually trigger the shutdown of socket. This solves the issue when kernel_sendmsg is currently blocked while a timeout occurs.

Signed-off-by Pranay Kr. Srivastava <[email protected]>

---
 drivers/block/nbd.c | 69 ++++++++++++++++++++++++++++++++++-------------------
 1 file changed, 44 insertions(+), 25 deletions(-)

drivers/block/nbd.c +1 -0

index 56f7f5d..586d946 100644
@@ -39,6 +39,7 @@
 #include <asm/types.h>
 
 #include <linux/nbd.h>
+#include <linux/workqueue.h>
 
 struct nbd_device {
 	u32 flags;

drivers/block/nbd.c +17 -20

+
 static inline struct device *nbd_to_dev(struct nbd_device *nbd)
 {
 	return disk_to_dev(nbd->disk);

@@ -172,39 +182,35 @@ static void nbd_end_request(struct nbd_device *nbd, struct request *req)
  */
 static void sock_shutdown(struct nbd_device *nbd)
 {
-	spin_lock_irq(&nbd->sock_lock);
-
-	if (!nbd->sock) {
-		spin_unlock_irq(&nbd->sock_lock);
-		return;
-	}
+	struct socket *sock;
 
-	dev_warn(disk_to_dev(nbd->disk), "shutting down socket\n");
-	kernel_sock_shutdown(nbd->sock, SHUT_RDWR);
-	sockfd_put(nbd->sock);
+	spin_lock(&nbd->sock_lock);
+	sock = nbd->sock;
 	nbd->sock = NULL;
-	spin_unlock_irq(&nbd->sock_lock);
+	spin_unlock(&nbd->sock_lock);
+
+	if (!sock)
+		return;
 
 	del_timer(&nbd->timeout_timer);
+	dev_warn(disk_to_dev(nbd->disk), "shutting down socket\n");
+	kernel_sock_shutdown(sock, SHUT_RDWR);
+	sockfd_put(sock);
 }
 
 static void nbd_xmit_timeout(unsigned long arg)
 {
 	struct nbd_device *nbd = (struct nbd_device *)arg;
-	unsigned long flags;
 
 	if (list_empty(&nbd->queue_head))
 		return;
-
-	spin_lock_irqsave(&nbd->sock_lock, flags);
-
 	nbd->timedout = true;
-
-	if (nbd->sock)
-		kernel_sock_shutdown(nbd->sock, SHUT_RDWR);
-
-	spin_unlock_irqrestore(&nbd->sock_lock, flags);
-
+	schedule_work(&nbd->ws_shutdown);
+	/*
+	 * Make sure sender thread sees nbd->timedout.
+	 */
+	smp_wmb();
+	wake_up(&nbd->waiting_wq);
 	dev_err(nbd_to_dev(nbd), "Connection timed out, shutting down connection\n");
 }
 

drivers/block/nbd.c +10 -1

 	}
 
 	nbd->task_send = NULL;

@@ -668,6 +678,7 @@ static void nbd_reset(struct nbd_device *nbd)
 	set_capacity(nbd->disk, 0);
 	nbd->flags = 0;
 	nbd->xmit_timeout = 0;
+	INIT_WORK(&nbd->ws_shutdown, nbd_ws_func_shutdown);
 	queue_flag_clear_unlocked(QUEUE_FLAG_DISCARD, nbd->disk->queue);
 	del_timer_sync(&nbd->timeout_timer);
 }

@@ -802,11 +813,11 @@ static int __nbd_ioctl(struct block_device *bdev, struct nbd_device *nbd,
 		error = nbd_thread_recv(nbd, bdev);
 		nbd_dev_dbg_close(nbd);
 		kthread_stop(thread);
+		sock_shutdown(nbd);
 
 		mutex_lock(&nbd->tx_lock);
 		nbd->task_recv = NULL;
 
-		sock_shutdown(nbd);
 		nbd_clear_que(nbd);
 		kill_bdev(bdev);
 		nbd_bdev_reset(bdev);

@@ -862,6 +873,14 @@ static const struct block_device_operations nbd_fops = {
 	.compat_ioctl =	nbd_ioctl,
 };
 
+static void nbd_ws_func_shutdown(struct work_struct *ws_nbd)
+{
+	struct nbd_device *nbd_dev = container_of(ws_nbd, struct nbd_device,
+			ws_shutdown);
+
+	sock_shutdown(nbd_dev);
+}
+
 #if IS_ENABLED(CONFIG_DEBUG_FS)
 
 static int nbd_dbg_tasks_show(struct seq_file *s, void *unused)

When a timeout occurs or a recv fails, then instead of abruplty killing nbd block device

wait for it's users to finish.

This is more required when filesystem(s) like ext2 or ext3 don't expect their buffer heads to disappear while the filesystem is mounted.

Each open of a nbd device is refcounted, while the userland program [nbd-client] doing the NBD_DO_IT ioctl would now wait for any other users of this device before invalidating the nbd device.

A timedout or a disconnected device, if in use, can't be used until it has been resetted. The resetting happens

when all tasks having this bdev open closes this bdev.

Signed-off-by Pranay Kr. Srivastava <[email protected]>

---
 drivers/block/nbd.c | 124 ++++++++++++++++++++++++++++++++++++++++------------
 1 file changed, 96 insertions(+), 28 deletions(-)

Quoted 2 lines · Pranay Kr. Srivastava

When a timeout occurs or a recv fails, then instead of abruplty killing nbd block device

s/abruplty/abruptly/

Quoted 1 line

wait for it's users to finish.

s/it's/its/

Quoted 10 lines

This is more required when filesystem(s) like ext2 or ext3 don't expect their buffer heads to disappear while the filesystem is mounted.

Each open of a nbd device is refcounted, while the userland program [nbd-client] doing the NBD_DO_IT ioctl would now wait for any other users of this device before invalidating the nbd device.

Quoted 2 lines

A timedout or a disconnected device, if in use, can't be used until it has been resetted. The resetting happens

s/resetted/reset/

Quoted 7 lines

when all tasks having this bdev open closes this bdev.

Signed-off-by Pranay Kr. Srivastava <[email protected]>

---
 drivers/block/nbd.c | 124 ++++++++++++++++++++++++++++++++++++++++------------
 1 file changed, 96 insertions(+), 28 deletions(-)

--
Eric Blake eblake redhat com +1-919-301-3266
Libvirt virtualization library http://libvirt.org

When a timeout occurs or a recv fails, then
instead of abruplty killing nbd block device
wait for it's users to finish.

This is more required when filesystem(s) like ext2 or ext3 don't expect their buffer heads to disappear while the filesystem is mounted.

Each open of a nbd device is refcounted, while the userland program [nbd-client] doing the NBD_DO_IT ioctl would now wait for any other users of this device before invalidating the nbd device.

A timedout or a disconnected device, if in use, can't be used until it has been resetted. The resetting happens when all tasks having this bdev open closes this bdev.

Signed-off-by Pranay Kr. Srivastava <[email protected]>

---
 drivers/block/nbd.c | 124 ++++++++++++++++++++++++++++++++++++++++------------
 1 file changed, 96 insertions(+), 28 deletions(-)

drivers/block/nbd.c +96 -29

index 9223b09..0587bbd 100644
@@ -70,10 +70,13 @@ struct nbd_device {
 #if IS_ENABLED(CONFIG_DEBUG_FS)
        struct dentry *dbg_dir;
 #endif
+
        /*
-       *This is specifically for calling sock_shutdown, for now.
-       */
+        *This is specifically for calling sock_shutdown, for now.
+        */
        struct work_struct ws_shutdown;
+       struct kref users;
+       struct completion user_completion;
 };
 
 #if IS_ENABLED(CONFIG_DEBUG_FS)

@@ -104,6 +107,8 @@ static DEFINE_SPINLOCK(nbd_lock);
  * Shutdown function for nbd_dev work struct.
  */
 static void nbd_ws_func_shutdown(struct work_struct *);
+static void nbd_kref_release(struct kref *);
+static int nbd_size_clear(struct nbd_device *, struct block_device *);
 
 static inline struct device *nbd_to_dev(struct nbd_device *nbd)
 {

@@ -129,7 +134,7 @@ static const char *nbdcmd_to_ascii(int cmd)
 
 static int nbd_size_clear(struct nbd_device *nbd, struct block_device *bdev)
 {
-       bdev->bd_inode->i_size = 0;
+       i_size_write(bdev->bd_inode, 0);
        set_capacity(nbd->disk, 0);
        kobject_uevent(&nbd_to_dev(nbd)->kobj, KOBJ_CHANGE);
 

@@ -141,7 +146,7 @@ static void nbd_size_update(struct nbd_device *nbd, struct block_device *bdev)
        if (!nbd_is_connected(nbd))
                return;
 
-       bdev->bd_inode->i_size = nbd->bytesize;
+       i_size_write(bdev->bd_inode, nbd->bytesize);
        set_capacity(nbd->disk, nbd->bytesize >> 9);
        kobject_uevent(&nbd_to_dev(nbd)->kobj, KOBJ_CHANGE);
 }

@@ -150,11 +155,9 @@ static int nbd_size_set(struct nbd_device *nbd, struct block_device *bdev,
                        int blocksize, int nr_blocks)
 {
        int ret;
-
        ret = set_blocksize(bdev, blocksize);
        if (ret)
                return ret;
-
        nbd->blksize = blocksize;
        nbd->bytesize = (loff_t)blocksize * (loff_t)nr_blocks;
 

@@ -202,14 +205,19 @@ static void nbd_xmit_timeout(unsigned long arg)
 {
        struct nbd_device *nbd = (struct nbd_device *)arg;
 
+       if (nbd->timedout)
+               return;
+
        if (list_empty(&nbd->queue_head))
                return;
+
        nbd->timedout = true;
-       schedule_work(&nbd->ws_shutdown);
+
        /*
         * Make sure sender thread sees nbd->timedout.
         */
        smp_wmb();
+       schedule_work(&nbd->ws_shutdown);
        wake_up(&nbd->waiting_wq);
        dev_err(nbd_to_dev(nbd), "Connection timed out, shutting down connection\n");
 }

@@ -476,8 +484,6 @@ static int nbd_thread_recv(struct nbd_device *nbd, struct block_device *bdev)
                nbd_end_request(nbd, req);
        }
 
-       nbd_size_clear(nbd, bdev);
-
        device_remove_file(disk_to_dev(nbd->disk), &dev_attr_pid);
 
        nbd->task_recv = NULL;

@@ -580,8 +586,8 @@ static int nbd_thread_send(void *data)
        while (!kthread_should_stop() || !list_empty(&nbd->waiting_queue)) {
                /* wait for something to do */
                wait_event_interruptible(nbd->waiting_wq,
-                               kthread_should_stop() ||
-                               !list_empty(&nbd->waiting_queue));
+                                        kthread_should_stop() ||
+                                        !list_empty(&nbd->waiting_queue));
 
                /* extract request */
                if (list_empty(&nbd->waiting_queue))

@@ -589,11 +595,11 @@ static int nbd_thread_send(void *data)
 
                spin_lock_irq(&nbd->queue_lock);
                req = list_entry(nbd->waiting_queue.next, struct request,
-                               queuelist);
+                                queuelist);
                list_del_init(&req->queuelist);
                spin_unlock_irq(&nbd->queue_lock);
 
-               nbd_handle_req(nbd, req);
+               /* handle request */
                if (nbd->timedout) {
                        req->errors++;
                        nbd_end_request(nbd, req);

@@ -654,12 +660,13 @@ static int nbd_set_socket(struct nbd_device *nbd, struct socket *sock)
        int ret = 0;
 
        spin_lock(&nbd->sock_lock);
-       if (nbd->sock)
+
+       if (nbd->sock || nbd->timedout)
                ret = -EBUSY;
        else
                nbd->sock = sock;
-       spin_unlock(&nbd->sock_lock);
 
+       spin_unlock(&nbd->sock_lock);
        return ret;
 }
 

@@ -674,6 +681,7 @@ static void nbd_reset(struct nbd_device *nbd)
        nbd->flags = 0;
        nbd->xmit_timeout = 0;
        INIT_WORK(&nbd->ws_shutdown, nbd_ws_func_shutdown);
+       init_completion(&nbd->user_completion);
        queue_flag_clear_unlocked(QUEUE_FLAG_DISCARD, nbd->disk->queue);
        del_timer_sync(&nbd->timeout_timer);
 }

@@ -708,6 +716,9 @@ static void nbd_dev_dbg_close(struct nbd_device *nbd);
 static int __nbd_ioctl(struct block_device *bdev, struct nbd_device *nbd,
                       unsigned int cmd, unsigned long arg)
 {
+       if (nbd->timedout || nbd->disconnect)
+               return -EBUSY;
+
        switch (cmd) {
        case NBD_DISCONNECT: {
                struct request sreq;

@@ -737,7 +748,6 @@ static int __nbd_ioctl(struct block_device *bdev, struct nbd_device *nbd,
                nbd_clear_que(nbd);
                BUG_ON(!list_empty(&nbd->queue_head));
                BUG_ON(!list_empty(&nbd->waiting_queue));
-               kill_bdev(bdev);
                return 0;
 
        case NBD_SET_SOCK: {

@@ -756,7 +766,6 @@ static int __nbd_ioctl(struct block_device *bdev, struct nbd_device *nbd,
 
        case NBD_SET_BLKSIZE: {
                loff_t bsize = div_s64(nbd->bytesize, arg);
-
                return nbd_size_set(nbd, bdev, arg, bsize);
        }
 

@@ -808,22 +817,29 @@ static int __nbd_ioctl(struct block_device *bdev, struct nbd_device *nbd,
                error = nbd_thread_recv(nbd, bdev);
                nbd_dev_dbg_close(nbd);
                kthread_stop(thread);
-               sock_shutdown(nbd);
 
-               mutex_lock(&nbd->tx_lock);
-               nbd->task_recv = NULL;
-
-               nbd_clear_que(nbd);
-               kill_bdev(bdev);
-               nbd_bdev_reset(bdev);
+               sock_shutdown(nbd);
 
                if (nbd->disconnect) /* user requested, ignore socket errors */
                        error = 0;
                if (nbd->timedout)
                        error = -ETIMEDOUT;
 
-               nbd_reset(nbd);
+               mutex_lock(&nbd->tx_lock);
+               nbd_clear_que(nbd);
+               nbd->disconnect = true; /* To kill bdev*/
+               mutex_unlock(&nbd->tx_lock);
+               cancel_work_sync(&nbd->ws_shutdown);
+               kref_put(&nbd->users, nbd_kref_release);
+               wait_for_completion(&nbd->user_completion);
 
+               mutex_lock(&bdev->bd_mutex);
+               if (!kref_get_unless_zero(&nbd->users))
+                       kref_init(&nbd->users);
+               mutex_unlock(&bdev->bd_mutex);
+
+               mutex_lock(&nbd->tx_lock);
+               nbd_reset(nbd);
                return error;
        }
 

@@ -861,19 +877,71 @@ static int nbd_ioctl(struct block_device *bdev, fmode_t mode,
 
        return error;
 }
+static void nbd_kref_release(struct kref *kref_users)
+{
+       struct nbd_device *nbd = container_of(kref_users, struct nbd_device,
+                       users);
+       schedule_work(&nbd->ws_shutdown);
+}
+
+static int nbd_open(struct block_device *bdev, fmode_t mode)
+{
+       struct nbd_device *nbd_dev = bdev->bd_disk->private_data;
+
+       if (!kref_get_unless_zero(&nbd_dev->users))
+               kref_init(&nbd_dev->users);
+
+       pr_debug("Opening nbd_dev %s. Active users = %u\n",
+                       bdev->bd_disk->disk_name,
+                       atomic_read(&nbd_dev->users.refcount)
+               );
+       return 0;
+}
+
+static void nbd_release(struct gendisk *disk, fmode_t mode)
+{
+       struct nbd_device *nbd_dev = disk->private_data;
+
+       kref_put(&nbd_dev->users,  nbd_kref_release);
+
+       pr_debug("Closing nbd_dev %s. Active users = %u\n",
+                       disk->disk_name,
+                       atomic_read(&nbd_dev->users.refcount)
+               );
+}
 
 static const struct block_device_operations nbd_fops = {
        .owner =        THIS_MODULE,
        .ioctl =        nbd_ioctl,
        .compat_ioctl = nbd_ioctl,
+       .open =         nbd_open,
+       .release =      nbd_release
 };
 
+
 static void nbd_ws_func_shutdown(struct work_struct *ws_nbd)
 {
        struct nbd_device *nbd_dev = container_of(ws_nbd, struct nbd_device,
-                       ws_shutdown);
-
-       sock_shutdown(nbd_dev);
+                                                       ws_shutdown);
+       struct block_device *bdev = bdget(part_devt(
+                                               dev_to_part(nbd_to_dev(nbd_dev))
+                                               )
+                                       );
+       BUG_ON(!bdev);
+       if (nbd_dev->timedout)
+               sock_shutdown(nbd_dev);
+
+       if (nbd_dev->disconnect) {
+               mutex_lock(&nbd_dev->tx_lock);
+               nbd_dev->task_recv = NULL;
+               nbd_clear_que(nbd_dev);
+               kill_bdev(bdev);
+               nbd_bdev_reset(bdev);
+               mutex_unlock(&nbd_dev->tx_lock);
+               nbd_size_clear(nbd_dev, bdev);
+               complete(&nbd_dev->user_completion);
+       }
+       bdput(bdev);
 }
 
 #if IS_ENABLED(CONFIG_DEBUG_FS)
--

1.9.1

When a timeout occurs or a recv fails, then
instead of abruplty killing nbd block device
wait for it's users to finish.

This is more required when filesystem(s) like ext2 or ext3 don't expect their buffer heads to disappear while the filesystem is mounted.

Each open of a nbd device is refcounted, while the userland program [nbd-client] doing the NBD_DO_IT ioctl would now wait for any other users of this device before invalidating the nbd device.

A timedout or a disconnected device, if in use, can't be used until it has been resetted. The resetting happens when all tasks having this bdev open closes this bdev.

Signed-off-by Pranay Kr. Srivastava <[email protected]>

---
 drivers/block/nbd.c | 124 ++++++++++++++++++++++++++++++++++++++++------------
 1 file changed, 96 insertions(+), 28 deletions(-)

drivers/block/nbd.c +96 -28

index 9223b09..0587bbd 100644
@@ -70,10 +70,13 @@ struct nbd_device {
 #if IS_ENABLED(CONFIG_DEBUG_FS)
 	struct dentry *dbg_dir;
 #endif
+
 	/*
-	*This is specifically for calling sock_shutdown, for now.
-	*/
+	 *This is specifically for calling sock_shutdown, for now.
+	 */
 	struct work_struct ws_shutdown;
+	struct kref users;
+	struct completion user_completion;
 };
 
 #if IS_ENABLED(CONFIG_DEBUG_FS)

@@ -104,6 +107,8 @@ static DEFINE_SPINLOCK(nbd_lock);
  * Shutdown function for nbd_dev work struct.
  */
 static void nbd_ws_func_shutdown(struct work_struct *);
+static void nbd_kref_release(struct kref *);
+static int nbd_size_clear(struct nbd_device *, struct block_device *);
 
 static inline struct device *nbd_to_dev(struct nbd_device *nbd)
 {

@@ -129,7 +134,7 @@ static const char *nbdcmd_to_ascii(int cmd)
 
 static int nbd_size_clear(struct nbd_device *nbd, struct block_device *bdev)
 {
-	bdev->bd_inode->i_size = 0;
+	i_size_write(bdev->bd_inode, 0);
 	set_capacity(nbd->disk, 0);
 	kobject_uevent(&nbd_to_dev(nbd)->kobj, KOBJ_CHANGE);
 

@@ -141,7 +146,7 @@ static void nbd_size_update(struct nbd_device *nbd, struct block_device *bdev)
 	if (!nbd_is_connected(nbd))
 		return;
 
-	bdev->bd_inode->i_size = nbd->bytesize;
+	i_size_write(bdev->bd_inode, nbd->bytesize);
 	set_capacity(nbd->disk, nbd->bytesize >> 9);
 	kobject_uevent(&nbd_to_dev(nbd)->kobj, KOBJ_CHANGE);
 }

@@ -150,11 +155,9 @@ static int nbd_size_set(struct nbd_device *nbd, struct block_device *bdev,
 			int blocksize, int nr_blocks)
 {
 	int ret;
-
 	ret = set_blocksize(bdev, blocksize);
 	if (ret)
 		return ret;
-
 	nbd->blksize = blocksize;
 	nbd->bytesize = (loff_t)blocksize * (loff_t)nr_blocks;
 

@@ -202,14 +205,19 @@ static void nbd_xmit_timeout(unsigned long arg)
 {
 	struct nbd_device *nbd = (struct nbd_device *)arg;
 
+	if (nbd->timedout)
+		return;
+
 	if (list_empty(&nbd->queue_head))
 		return;
+
 	nbd->timedout = true;
-	schedule_work(&nbd->ws_shutdown);
+
 	/*
 	 * Make sure sender thread sees nbd->timedout.
 	 */
 	smp_wmb();
+	schedule_work(&nbd->ws_shutdown);
 	wake_up(&nbd->waiting_wq);
 	dev_err(nbd_to_dev(nbd), "Connection timed out, shutting down connection\n");
 }

@@ -476,8 +484,6 @@ static int nbd_thread_recv(struct nbd_device *nbd, struct block_device *bdev)
 		nbd_end_request(nbd, req);
 	}
 
-	nbd_size_clear(nbd, bdev);
-
 	device_remove_file(disk_to_dev(nbd->disk), &dev_attr_pid);
 
 	nbd->task_recv = NULL;

@@ -580,8 +586,8 @@ static int nbd_thread_send(void *data)
 	while (!kthread_should_stop() || !list_empty(&nbd->waiting_queue)) {
 		/* wait for something to do */
 		wait_event_interruptible(nbd->waiting_wq,
-				kthread_should_stop() ||
-				!list_empty(&nbd->waiting_queue));
+					 kthread_should_stop() ||
+					 !list_empty(&nbd->waiting_queue));
 
 		/* extract request */
 		if (list_empty(&nbd->waiting_queue))

@@ -589,11 +595,11 @@ static int nbd_thread_send(void *data)
 
 		spin_lock_irq(&nbd->queue_lock);
 		req = list_entry(nbd->waiting_queue.next, struct request,
-				queuelist);
+				 queuelist);
 		list_del_init(&req->queuelist);
 		spin_unlock_irq(&nbd->queue_lock);
 
-		nbd_handle_req(nbd, req);
+		/* handle request */
 		if (nbd->timedout) {
 			req->errors++;
 			nbd_end_request(nbd, req);

@@ -654,12 +660,13 @@ static int nbd_set_socket(struct nbd_device *nbd, struct socket *sock)
 	int ret = 0;
 
 	spin_lock(&nbd->sock_lock);
-	if (nbd->sock)
+
+	if (nbd->sock || nbd->timedout)
 		ret = -EBUSY;
 	else
 		nbd->sock = sock;
-	spin_unlock(&nbd->sock_lock);
 
+	spin_unlock(&nbd->sock_lock);
 	return ret;
 }
 

@@ -674,6 +681,7 @@ static void nbd_reset(struct nbd_device *nbd)
 	nbd->flags = 0;
 	nbd->xmit_timeout = 0;
 	INIT_WORK(&nbd->ws_shutdown, nbd_ws_func_shutdown);
+	init_completion(&nbd->user_completion);
 	queue_flag_clear_unlocked(QUEUE_FLAG_DISCARD, nbd->disk->queue);
 	del_timer_sync(&nbd->timeout_timer);
 }

@@ -708,6 +716,9 @@ static void nbd_dev_dbg_close(struct nbd_device *nbd);
 static int __nbd_ioctl(struct block_device *bdev, struct nbd_device *nbd,
 		       unsigned int cmd, unsigned long arg)
 {
+	if (nbd->timedout || nbd->disconnect)
+		return -EBUSY;
+
 	switch (cmd) {
 	case NBD_DISCONNECT: {
 		struct request sreq;

@@ -737,7 +748,6 @@ static int __nbd_ioctl(struct block_device *bdev, struct nbd_device *nbd,
 		nbd_clear_que(nbd);
 		BUG_ON(!list_empty(&nbd->queue_head));
 		BUG_ON(!list_empty(&nbd->waiting_queue));
-		kill_bdev(bdev);
 		return 0;
 
 	case NBD_SET_SOCK: {

@@ -756,7 +766,6 @@ static int __nbd_ioctl(struct block_device *bdev, struct nbd_device *nbd,
 
 	case NBD_SET_BLKSIZE: {
 		loff_t bsize = div_s64(nbd->bytesize, arg);
-
 		return nbd_size_set(nbd, bdev, arg, bsize);
 	}
 

@@ -808,22 +817,29 @@ static int __nbd_ioctl(struct block_device *bdev, struct nbd_device *nbd,
 		error = nbd_thread_recv(nbd, bdev);
 		nbd_dev_dbg_close(nbd);
 		kthread_stop(thread);
-		sock_shutdown(nbd);
 
-		mutex_lock(&nbd->tx_lock);
-		nbd->task_recv = NULL;
-
-		nbd_clear_que(nbd);
-		kill_bdev(bdev);
-		nbd_bdev_reset(bdev);
+		sock_shutdown(nbd);
 
 		if (nbd->disconnect) /* user requested, ignore socket errors */
 			error = 0;
 		if (nbd->timedout)
 			error = -ETIMEDOUT;
 
-		nbd_reset(nbd);
+		mutex_lock(&nbd->tx_lock);
+		nbd_clear_que(nbd);
+		nbd->disconnect = true; /* To kill bdev*/
+		mutex_unlock(&nbd->tx_lock);
+		cancel_work_sync(&nbd->ws_shutdown);
+		kref_put(&nbd->users, nbd_kref_release);
+		wait_for_completion(&nbd->user_completion);
 
+		mutex_lock(&bdev->bd_mutex);
+		if (!kref_get_unless_zero(&nbd->users))
+			kref_init(&nbd->users);
+		mutex_unlock(&bdev->bd_mutex);
+
+		mutex_lock(&nbd->tx_lock);
+		nbd_reset(nbd);
 		return error;
 	}
 

@@ -861,19 +877,71 @@ static int nbd_ioctl(struct block_device *bdev, fmode_t mode,
 
 	return error;
 }
+static void nbd_kref_release(struct kref *kref_users)
+{
+	struct nbd_device *nbd = container_of(kref_users, struct nbd_device,
+			users);
+	schedule_work(&nbd->ws_shutdown);
+}
+
+static int nbd_open(struct block_device *bdev, fmode_t mode)
+{
+	struct nbd_device *nbd_dev = bdev->bd_disk->private_data;
+
+	if (!kref_get_unless_zero(&nbd_dev->users))
+		kref_init(&nbd_dev->users);
+
+	pr_debug("Opening nbd_dev %s. Active users = %u\n",
+			bdev->bd_disk->disk_name,
+			atomic_read(&nbd_dev->users.refcount)
+		);
+	return 0;
+}
+
+static void nbd_release(struct gendisk *disk, fmode_t mode)
+{
+	struct nbd_device *nbd_dev = disk->private_data;
+
+	kref_put(&nbd_dev->users,  nbd_kref_release);
+
+	pr_debug("Closing nbd_dev %s. Active users = %u\n",
+			disk->disk_name,
+			atomic_read(&nbd_dev->users.refcount)
+		);
+}
 
 static const struct block_device_operations nbd_fops = {
 	.owner =	THIS_MODULE,
 	.ioctl =	nbd_ioctl,
 	.compat_ioctl =	nbd_ioctl,
+	.open =		nbd_open,
+	.release =	nbd_release
 };
 
+
 static void nbd_ws_func_shutdown(struct work_struct *ws_nbd)
 {
 	struct nbd_device *nbd_dev = container_of(ws_nbd, struct nbd_device,
-			ws_shutdown);
-
-	sock_shutdown(nbd_dev);
+							ws_shutdown);
+	struct block_device *bdev = bdget(part_devt(
+						dev_to_part(nbd_to_dev(nbd_dev))
+						)
+					);
+	BUG_ON(!bdev);
+	if (nbd_dev->timedout)
+		sock_shutdown(nbd_dev);
+
+	if (nbd_dev->disconnect) {
+		mutex_lock(&nbd_dev->tx_lock);
+		nbd_dev->task_recv = NULL;
+		nbd_clear_que(nbd_dev);
+		kill_bdev(bdev);
+		nbd_bdev_reset(bdev);
+		mutex_unlock(&nbd_dev->tx_lock);
+		nbd_size_clear(nbd_dev, bdev);
+		complete(&nbd_dev->user_completion);
+	}
+	bdput(bdev);
 }
 
 #if IS_ENABLED(CONFIG_DEBUG_FS)

Hi,

Quoted 13 lines · On Friday 24 June 2016 13:09:36 Pranay Kr. Srivastava wrote:

When a timeout occurs or a recv fails, then
instead of abruplty killing nbd block device
wait for it's users to finish.

This is more required when filesystem(s) like ext2 or ext3 don't expect their buffer heads to disappear while the filesystem is mounted.

Each open of a nbd device is refcounted, while the userland program [nbd-client] doing the NBD_DO_IT ioctl would now wait for any other users of this device before invalidating the nbd device.

Quoted 3 lines

A timedout or a disconnected device, if in use, can't be used until it has been resetted. The resetting happens when all tasks having this bdev open closes this bdev.

Sorry, but this patch is unreadable. You are changing so many unrelated whitespaces, lines, comments (that you introduced yourself in a previous patch) and unrelated code. Please keep only the things that are

necessary for this single patch. Everything else can go into different patches. Also it would be good to run checkpatch sometimes.

Also using your own atomic implementation instead of kref would be good perhaps. Although I thought kref would work at the beginning but it seems not to.

Best Regards,

Markus

Quoted 274 lines

Signed-off-by Pranay Kr. Srivastava <[email protected]>

---
 drivers/block/nbd.c | 124 ++++++++++++++++++++++++++++++++++++++++------------
 1 file changed, 96 insertions(+), 28 deletions(-)

drivers/block/nbd.c +89 -28

index 9223b09..0587bbd 100644
@@ -70,10 +70,13 @@ struct nbd_device {
 #if IS_ENABLED(CONFIG_DEBUG_FS)
      struct dentry *dbg_dir;
 #endif
+
      /*
-     *This is specifically for calling sock_shutdown, for now.
-     */
+      *This is specifically for calling sock_shutdown, for now.
+      */
      struct work_struct ws_shutdown;
+     struct kref users;
+     struct completion user_completion;
 };
 
 #if IS_ENABLED(CONFIG_DEBUG_FS)

@@ -104,6 +107,8 @@ static DEFINE_SPINLOCK(nbd_lock);
  * Shutdown function for nbd_dev work struct.
  */
 static void nbd_ws_func_shutdown(struct work_struct *);
+static void nbd_kref_release(struct kref *);
+static int nbd_size_clear(struct nbd_device *, struct block_device *);
 
 static inline struct device *nbd_to_dev(struct nbd_device *nbd)
 {

@@ -129,7 +134,7 @@ static const char *nbdcmd_to_ascii(int cmd)
 
 static int nbd_size_clear(struct nbd_device *nbd, struct block_device *bdev)
 {
-     bdev->bd_inode->i_size = 0;
+     i_size_write(bdev->bd_inode, 0);
      set_capacity(nbd->disk, 0);
      kobject_uevent(&nbd_to_dev(nbd)->kobj, KOBJ_CHANGE);
 

@@ -141,7 +146,7 @@ static void nbd_size_update(struct nbd_device *nbd, struct block_device *bdev)
      if (!nbd_is_connected(nbd))
              return;
 
-     bdev->bd_inode->i_size = nbd->bytesize;
+     i_size_write(bdev->bd_inode, nbd->bytesize);
      set_capacity(nbd->disk, nbd->bytesize >> 9);
      kobject_uevent(&nbd_to_dev(nbd)->kobj, KOBJ_CHANGE);
 }

@@ -150,11 +155,9 @@ static int nbd_size_set(struct nbd_device *nbd, struct block_device *bdev,
                      int blocksize, int nr_blocks)
 {
      int ret;
-
      ret = set_blocksize(bdev, blocksize);
      if (ret)
              return ret;
-
      nbd->blksize = blocksize;
      nbd->bytesize = (loff_t)blocksize * (loff_t)nr_blocks;
 

@@ -202,14 +205,19 @@ static void nbd_xmit_timeout(unsigned long arg)
 {
      struct nbd_device *nbd = (struct nbd_device *)arg;
 
+     if (nbd->timedout)
+             return;
+
      if (list_empty(&nbd->queue_head))
              return;
+
      nbd->timedout = true;
-     schedule_work(&nbd->ws_shutdown);
+
      /*
       * Make sure sender thread sees nbd->timedout.
       */
      smp_wmb();
+     schedule_work(&nbd->ws_shutdown);
      wake_up(&nbd->waiting_wq);
      dev_err(nbd_to_dev(nbd), "Connection timed out, shutting down connection\n");
 }

@@ -476,8 +484,6 @@ static int nbd_thread_recv(struct nbd_device *nbd, struct block_device *bdev)
              nbd_end_request(nbd, req);
      }
 
-     nbd_size_clear(nbd, bdev);
-
      device_remove_file(disk_to_dev(nbd->disk), &dev_attr_pid);
 
      nbd->task_recv = NULL;

@@ -580,8 +586,8 @@ static int nbd_thread_send(void *data)
      while (!kthread_should_stop() || !list_empty(&nbd->waiting_queue)) {
              /* wait for something to do */
              wait_event_interruptible(nbd->waiting_wq,
-                             kthread_should_stop() ||
-                             !list_empty(&nbd->waiting_queue));
+                                      kthread_should_stop() ||
+                                      !list_empty(&nbd->waiting_queue));
 
              /* extract request */
              if (list_empty(&nbd->waiting_queue))

@@ -589,11 +595,11 @@ static int nbd_thread_send(void *data)
 
              spin_lock_irq(&nbd->queue_lock);
              req = list_entry(nbd->waiting_queue.next, struct request,
-                             queuelist);
+                              queuelist);
              list_del_init(&req->queuelist);
              spin_unlock_irq(&nbd->queue_lock);
 
-             nbd_handle_req(nbd, req);
+             /* handle request */
              if (nbd->timedout) {
                      req->errors++;
                      nbd_end_request(nbd, req);

@@ -654,12 +660,13 @@ static int nbd_set_socket(struct nbd_device *nbd, struct socket *sock)
      int ret = 0;
 
      spin_lock(&nbd->sock_lock);
-     if (nbd->sock)
+
+     if (nbd->sock || nbd->timedout)
              ret = -EBUSY;
      else
              nbd->sock = sock;
-     spin_unlock(&nbd->sock_lock);
 
+     spin_unlock(&nbd->sock_lock);
      return ret;
 }
 

@@ -674,6 +681,7 @@ static void nbd_reset(struct nbd_device *nbd)
      nbd->flags = 0;
      nbd->xmit_timeout = 0;
      INIT_WORK(&nbd->ws_shutdown, nbd_ws_func_shutdown);
+     init_completion(&nbd->user_completion);
      queue_flag_clear_unlocked(QUEUE_FLAG_DISCARD, nbd->disk->queue);
      del_timer_sync(&nbd->timeout_timer);
 }

@@ -708,6 +716,9 @@ static void nbd_dev_dbg_close(struct nbd_device *nbd);
 static int __nbd_ioctl(struct block_device *bdev, struct nbd_device *nbd,
                     unsigned int cmd, unsigned long arg)
 {
+     if (nbd->timedout || nbd->disconnect)
+             return -EBUSY;
+
      switch (cmd) {
      case NBD_DISCONNECT: {
              struct request sreq;

@@ -737,7 +748,6 @@ static int __nbd_ioctl(struct block_device *bdev, struct nbd_device *nbd,
              nbd_clear_que(nbd);
              BUG_ON(!list_empty(&nbd->queue_head));
              BUG_ON(!list_empty(&nbd->waiting_queue));
-             kill_bdev(bdev);
              return 0;
 
      case NBD_SET_SOCK: {

@@ -756,7 +766,6 @@ static int __nbd_ioctl(struct block_device *bdev, struct nbd_device *nbd,
 
      case NBD_SET_BLKSIZE: {
              loff_t bsize = div_s64(nbd->bytesize, arg);
-
              return nbd_size_set(nbd, bdev, arg, bsize);
      }
 

@@ -808,22 +817,29 @@ static int __nbd_ioctl(struct block_device *bdev, struct nbd_device *nbd,
              error = nbd_thread_recv(nbd, bdev);
              nbd_dev_dbg_close(nbd);
              kthread_stop(thread);
-             sock_shutdown(nbd);
 
-             mutex_lock(&nbd->tx_lock);
-             nbd->task_recv = NULL;
-
-             nbd_clear_que(nbd);
-             kill_bdev(bdev);
-             nbd_bdev_reset(bdev);
+             sock_shutdown(nbd);
 
              if (nbd->disconnect) /* user requested, ignore socket errors */
                      error = 0;
              if (nbd->timedout)
                      error = -ETIMEDOUT;
 
-             nbd_reset(nbd);
+             mutex_lock(&nbd->tx_lock);
+             nbd_clear_que(nbd);
+             nbd->disconnect = true; /* To kill bdev*/
+             mutex_unlock(&nbd->tx_lock);
+             cancel_work_sync(&nbd->ws_shutdown);
+             kref_put(&nbd->users, nbd_kref_release);
+             wait_for_completion(&nbd->user_completion);
 
+             mutex_lock(&bdev->bd_mutex);
+             if (!kref_get_unless_zero(&nbd->users))
+                     kref_init(&nbd->users);
+             mutex_unlock(&bdev->bd_mutex);
+
+             mutex_lock(&nbd->tx_lock);
+             nbd_reset(nbd);
              return error;
      }
 

@@ -861,19 +877,71 @@ static int nbd_ioctl(struct block_device *bdev, fmode_t mode,
 
      return error;
 }
+static void nbd_kref_release(struct kref *kref_users)
+{
+     struct nbd_device *nbd = container_of(kref_users, struct nbd_device,
+                     users);
+     schedule_work(&nbd->ws_shutdown);
+}
+
+static int nbd_open(struct block_device *bdev, fmode_t mode)
+{
+     struct nbd_device *nbd_dev = bdev->bd_disk->private_data;
+
+     if (!kref_get_unless_zero(&nbd_dev->users))
+             kref_init(&nbd_dev->users);
+
+     pr_debug("Opening nbd_dev %s. Active users = %u\n",
+                     bdev->bd_disk->disk_name,
+                     atomic_read(&nbd_dev->users.refcount)
+             );
+     return 0;
+}
+
+static void nbd_release(struct gendisk *disk, fmode_t mode)
+{
+     struct nbd_device *nbd_dev = disk->private_data;
+
+     kref_put(&nbd_dev->users,  nbd_kref_release);
+
+     pr_debug("Closing nbd_dev %s. Active users = %u\n",
+                     disk->disk_name,
+                     atomic_read(&nbd_dev->users.refcount)
+             );
+}
 
 static const struct block_device_operations nbd_fops = {
      .owner =        THIS_MODULE,
      .ioctl =        nbd_ioctl,
      .compat_ioctl = nbd_ioctl,
+     .open =         nbd_open,
+     .release =      nbd_release
 };
 
+
 static void nbd_ws_func_shutdown(struct work_struct *ws_nbd)
 {
      struct nbd_device *nbd_dev = container_of(ws_nbd, struct nbd_device,
-                     ws_shutdown);
-
-     sock_shutdown(nbd_dev);
+                                                     ws_shutdown);
+     struct block_device *bdev = bdget(part_devt(
+                                             dev_to_part(nbd_to_dev(nbd_dev))
+                                             )
+                                     );
+     BUG_ON(!bdev);
+     if (nbd_dev->timedout)
+             sock_shutdown(nbd_dev);
+
+     if (nbd_dev->disconnect) {
+             mutex_lock(&nbd_dev->tx_lock);
+             nbd_dev->task_recv = NULL;
+             nbd_clear_que(nbd_dev);

drivers/block/nbd.c static int nbd_ioctl(struct block_device *bdev, fmode_t mode, · line 947

Quoted 10 lines

drivers/block/nbd.c +7 -0

+             kill_bdev(bdev);
+             nbd_bdev_reset(bdev);
+             mutex_unlock(&nbd_dev->tx_lock);
+             nbd_size_clear(nbd_dev, bdev);
+             complete(&nbd_dev->user_completion);
+     }
+     bdput(bdev);
 }
 
 #if IS_ENABLED(CONFIG_DEBUG_FS)

--

Pengutronix e.K.                           |                             |
Industrial Linux Solutions                 | http://www.pengutronix.de/  |
Peiner Str. 6-8, 31137 Hildesheim, Germany | Phone: +49-5121-206917-0    |
Amtsgericht Hildesheim, HRA 2686           | Fax:   +49-5121-206917-5555 |