nbd: nbd fixes

From: "Pranay Kr. Srivastava" <[email protected]>

This patch
1) uses spin_lock instead of irq version.

2) removes the goto statement in case a socket is already assigned with simple if-else statement.

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

---
 drivers/block/nbd.c | 15 +++++----------
 1 file changed, 5 insertions(+), 10 deletions(-)

drivers/block/nbd.c +5 -11

index 56f7f5d..766c401 100644
@@ -643,17 +643,12 @@ static int nbd_set_socket(struct nbd_device *nbd, struct socket *sock)
 {
        int ret = 0;
 
-       spin_lock_irq(&nbd->sock_lock);
-
-       if (nbd->sock) {
+       spin_lock(&nbd->sock_lock);
+       if (nbd->sock)
                ret = -EBUSY;
-               goto out;
-       }
-
-       nbd->sock = sock;
-
-out:
-       spin_unlock_irq(&nbd->sock_lock);
+       else
+               nbd->sock = sock;
+       spin_unlock(&nbd->sock_lock);
 
        return ret;
 }
--

1.9.1

Quoted 41 lines · Pranay Kr. Srivastava

From: "Pranay Kr. Srivastava" <[email protected]>

This patch
1) uses spin_lock instead of irq version.

2) removes the goto statement in case a socket is already assigned with simple if-else statement.

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

---
 drivers/block/nbd.c | 15 +++++----------
 1 file changed, 5 insertions(+), 10 deletions(-)

drivers/block/nbd.c +5 -11

index 56f7f5d..766c401 100644
@@ -643,17 +643,12 @@ static int nbd_set_socket(struct nbd_device *nbd, struct socket *sock)
 {
        int ret = 0;
 
-       spin_lock_irq(&nbd->sock_lock);
-
-       if (nbd->sock) {
+       spin_lock(&nbd->sock_lock);
+       if (nbd->sock)
                ret = -EBUSY;
-               goto out;
-       }
-
-       nbd->sock = sock;
-
-out:
-       spin_unlock_irq(&nbd->sock_lock);
+       else
+               nbd->sock = sock;
+       spin_unlock(&nbd->sock_lock);
 
        return ret;
 }
--

Quoted 2 lines

1.9.1

Hi Markus,

Did you get a chance to review V4 of this 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 | 57 +++++++++++++++++++++++++++++++++--------------------
 1 file changed, 36 insertions(+), 21 deletions(-)

drivers/block/nbd.c +36 -22

index 766c401..e362d44 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,8 @@ 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 +98,8 @@ static int max_part;
  */
 static DEFINE_SPINLOCK(nbd_lock);
 
+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 +177,36 @@ 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);
-
+       /*
+        * 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");
 }
 

@@ -588,7 +590,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;

@@ -663,6 +669,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);
 }

@@ -797,11 +804,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);

@@ -857,6 +864,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 | 57

+++++++++++++++++++++++++++++++++-------------------- 1 file changed, 36 insertions(+), 21 deletions(-)

drivers/block/nbd.c +3 -0

index 766c401..e362d44 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,8 @@ 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)

drivers/block/nbd.c +0 -0

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

@@ -172,39 +177,36 @@ static void nbd_end_request(struct nbd_device *nbd,

struct request *req) */
static void sock_shutdown(struct nbd_device *nbd) {

drivers/block/nbd.c +16 -19

-	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);
-
+	/*
+	 * 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"); }

drivers/block/nbd.c +10 -1

 	}
 
 	nbd->task_send = NULL;

@@ -663,6 +669,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);
 }

@@ -797,11 +804,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);

@@ -857,6 +864,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 50 lines · Markus Pargmann

Hi,

Quoted 47 lines · On 2016 M06 30, Thu 14:02:02 CEST 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.

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

---
 drivers/block/nbd.c | 57

+++++++++++++++++++++++++++++++++-------------------- 1 file changed, 36 insertions(+), 21 deletions(-)

drivers/block/nbd.c +3 -0

index 766c401..e362d44 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,8 @@ 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 +98,8 @@ static int max_part;
  */
 static DEFINE_SPINLOCK(nbd_lock);
 

Quoted 6 lines

drivers/block/nbd.c static int max_part; · line 102

Quoted 2 lines

drivers/block/nbd.c +2 -0

+static void nbd_ws_func_shutdown(struct work_struct *);
+

are you reading all the comments I had?...

At least respond to my comments if you disagree. I still can't see the

benefit

Quoted 2 lines

of a function signature here if we can avoid it.

That would require some code to be moved. So to avoid those unnecessary changes it was better to have a prototype.

It would've pissed you off more if I had tried
to get rid of protoype.

Quoted 63 lines

Quoted 63 lines

drivers/block/nbd.c +0 -0

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

@@ -172,39 +177,36 @@ static void nbd_end_request(struct nbd_device *nbd,

struct request *req) */
static void sock_shutdown(struct nbd_device *nbd) {

drivers/block/nbd.c +16 -19

-     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);
-
+     /*
+      * 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"); }

drivers/block/nbd.c +0 -0

@@ -588,7 +590,11 @@ static int nbd_thread_send(void *data)
              spin_unlock_irq(&nbd->queue_lock);
 

Quoted 9 lines

drivers/block/nbd.c static int nbd_thread_send(void *data) · line 597

Quoted 7 lines

drivers/block/nbd.c +5 -1

              /* handle request */
-             nbd_handle_req(nbd, req);
+             if (nbd->timedout) {
+                     req->errors++;
+                     nbd_end_request(nbd, req);
+             } else
+                     nbd_handle_req(nbd, req);

I already commented on this in the last patch. This is unrelated to the

patch.

Quoted 3 lines

If you disagree then please tell me why instead of sending the same thing again.

After trigerring worker thread its
not necessary that socket shutdown
actually was called before we handled
a request.

So the error would come in
actually later probably.

So i just wanted to avoid a longer
path for error to be thrown up.
Do correct me if this cant happen.

Quoted 2 lines

Also brackets on the else part would be preferred.

It might trigger checkpatch warning
but I am not 100% sure.

Quoted 22 lines

Regards,

Markus

Quoted 17 lines

drivers/block/nbd.c +2 -0

      }
 
      nbd->task_send = NULL;

@@ -663,6 +669,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);
 }

@@ -797,11 +804,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);
 

Quoted 8 lines

drivers/block/nbd.c static const struct block_device_operations

Quoted 8 lines

drivers/block/nbd.c +0 -1

              mutex_lock(&nbd->tx_lock);
              nbd->task_recv = NULL;
 
-             sock_shutdown(nbd);
              nbd_clear_que(nbd);
              kill_bdev(bdev);
              nbd_bdev_reset(bdev);

@@ -857,6 +864,14 @@ static const struct block_device_operations

nbd_fops =

Quoted 6 lines

drivers/block/nbd.c static const struct block_device_operations nbd_fops = {

Quoted 6 lines

drivers/block/nbd.c +3 -0

 { .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,

Quoted 11 lines

Quoted 8 lines

drivers/block/nbd.c +5 -0

+                     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 58 lines · On Sunday 10 July 2016 21:03:05 Pranay Srivastava wrote:

Quoted 50 lines · Markus Pargmann

Hi,

Quoted 47 lines · On 2016 M06 30, Thu 14:02:02 CEST 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.

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

---
 drivers/block/nbd.c | 57

+++++++++++++++++++++++++++++++++-------------------- 1 file changed, 36 insertions(+), 21 deletions(-)

drivers/block/nbd.c +3 -0

index 766c401..e362d44 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,8 @@ 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 +98,8 @@ static int max_part;
  */
 static DEFINE_SPINLOCK(nbd_lock);
 

Quoted 6 lines

drivers/block/nbd.c static int max_part; · line 102

Quoted 2 lines

drivers/block/nbd.c +2 -0

+static void nbd_ws_func_shutdown(struct work_struct *);
+

are you reading all the comments I had?...

At least respond to my comments if you disagree. I still can't see the

benefit

Quoted 8 lines

Quoted 2 lines

of a function signature here if we can avoid it.

That would require some code to be moved. So to avoid those unnecessary changes it was better to have a prototype.

It would've pissed you off more if I had tried
to get rid of protoype.

Ah I see, thanks.

Quoted 82 lines

Quoted 63 lines

Quoted 63 lines

drivers/block/nbd.c +0 -0

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

@@ -172,39 +177,36 @@ static void nbd_end_request(struct nbd_device *nbd,

struct request *req) */
static void sock_shutdown(struct nbd_device *nbd) {

drivers/block/nbd.c +16 -19

-     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);
-
+     /*
+      * 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"); }

drivers/block/nbd.c +0 -0

@@ -588,7 +590,11 @@ static int nbd_thread_send(void *data)
              spin_unlock_irq(&nbd->queue_lock);
 

Quoted 9 lines

drivers/block/nbd.c static int nbd_thread_send(void *data) · line 597

Quoted 7 lines

drivers/block/nbd.c +5 -1

              /* handle request */
-             nbd_handle_req(nbd, req);
+             if (nbd->timedout) {
+                     req->errors++;
+                     nbd_end_request(nbd, req);
+             } else
+                     nbd_handle_req(nbd, req);

I already commented on this in the last patch. This is unrelated to the

patch.

Quoted 3 lines

If you disagree then please tell me why instead of sending the same thing again.

After trigerring worker thread its
not necessary that socket shutdown
actually was called before we handled
a request.

Quoted 6 lines

So the error would come in
actually later probably.

So i just wanted to avoid a longer
path for error to be thrown up.
Do correct me if this cant happen.

Yes socket shutdown may not have been called when we reach this error handling code. But the timeout timer is usually in the range of seconds. I would assume that the time between triggering the worker and socket shutdown is within a few milliseconds. We would need to hit exactly this condition which would require a new request to be present.

I think this is very unlikely and it would be fine if we have a longer error path there. Am I missing something?

Also I just noticed that wake_up(&nbd->waiting_wq) in nbd_xmit_timeout() may not be necessary. In nbd_thread_send():

        wait_event_interruptible(nbd->waiting_wq,
                                 kthread_should_stop() ||
                                 !list_empty(&nbd->waiting_queue));

        if (list_empty(&nbd->waiting_queue))
                continue;

So wouldn't this wake_up() call simply result in nothing?
As soon as sock_shutdown() was called, the receiver thread would exit and close down nbd_thread_send() as well because of kthread_should_stop().

Quoted 5 lines

Quoted 2 lines

Also brackets on the else part would be preferred.

It might trigger checkpatch warning
but I am not 100% sure.

Documentation/CodingStyle documents this. See line 168.

Best Regards,

Markus

Quoted 31 lines

Quoted 22 lines

Regards,

Markus

Quoted 17 lines

drivers/block/nbd.c +2 -0

      }
 
      nbd->task_send = NULL;

@@ -663,6 +669,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);
 }

@@ -797,11 +804,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);
 

Quoted 8 lines

drivers/block/nbd.c static const struct block_device_operations

Quoted 8 lines

drivers/block/nbd.c +0 -1

              mutex_lock(&nbd->tx_lock);
              nbd->task_recv = NULL;
 
-             sock_shutdown(nbd);
              nbd_clear_que(nbd);
              kill_bdev(bdev);
              nbd_bdev_reset(bdev);

@@ -857,6 +864,14 @@ static const struct block_device_operations

nbd_fops =

Quoted 20 lines

Quoted 6 lines

drivers/block/nbd.c static const struct block_device_operations nbd_fops = {

Quoted 6 lines

drivers/block/nbd.c +3 -0

 { .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,

Quoted 11 lines

Quoted 8 lines

drivers/block/nbd.c +5 -0

+                     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 |

From: "Pranay Kr. Srivastava" <[email protected]>

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.

v5)
removed unnecessary code as per review of v4).

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

---
 drivers/block/nbd.c | 53 +++++++++++++++++++++++++++++++----------------------
 1 file changed, 31 insertions(+), 22 deletions(-)

drivers/block/nbd.c +31 -23

index 766c401..4919760 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,30 @@ 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);
        dev_err(nbd_to_dev(nbd), "Connection timed out, shutting down connection\n");
 }
 

@@ -663,6 +664,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);
 }

@@ -798,10 +800,9 @@ static int __nbd_ioctl(struct block_device *bdev, struct nbd_device *nbd,
                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);

@@ -857,6 +858,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

When a timeout occurs or a recv fails, then instead of abruplty killing nbd block device wait for its 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 reset happens when all tasks having this bdev open closes this bdev.

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

---
 drivers/block/nbd.c | 106

++++++++++++++++++++++++++++++++++++++++++---------- 1 file changed, 87 insertions(+), 19 deletions(-)

drivers/block/nbd.c +2 -0

index e362d44..fb56dd2 100644
@@ -72,6 +72,8 @@ struct nbd_device {
 #endif
 	/* 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)

drivers/block/nbd.c +0 -2

 	if (list_empty(&nbd->queue_head))
 		return;
 

@@ -472,8 +477,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;

drivers/block/nbd.c +4 -1

 	return ret;
 }
 

@@ -670,6 +674,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);
 }

@@ -704,6 +709,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;

@@ -733,7 +741,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: {

drivers/block/nbd.c +2 -8

 		return nbd_size_set(nbd, bdev, arg, bsize);
 	}

@@ -804,22 +810,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);

drivers/block/nbd.c +4 -0

+		mutex_unlock(&bdev->bd_mutex);
+
+		mutex_lock(&nbd->tx_lock);
+		nbd_reset(nbd);
 		return error;
 	}
 

drivers/block/nbd.c +27 -0

+}
+
+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)
+		);
+}

drivers/block/nbd.c +12 -0

+
+	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)

Quoted 39 lines · On 2016 M06 30, Thu 14:02:03 CEST Pranay Kr. Srivastava wrote:

When a timeout occurs or a recv fails, then instead of abruplty killing nbd block device wait for its 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 reset happens when all tasks having this bdev open closes this bdev.

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

---
 drivers/block/nbd.c | 106

++++++++++++++++++++++++++++++++++++++++++---------- 1 file changed, 87 insertions(+), 19 deletions(-)

drivers/block/nbd.c +2 -0

index e362d44..fb56dd2 100644
@@ -72,6 +72,8 @@ struct nbd_device {
 #endif
      /* 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)

@@ -99,6 +101,8 @@ static int max_part;
 static DEFINE_SPINLOCK(nbd_lock);
 

drivers/block/nbd.c static int max_part; · line 105

Quoted 3 lines

drivers/block/nbd.c +2 -0

 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 *);

More function signatures. Why?

drivers/block/nbd.c static int nbd_size_set(struct nbd_device *nbd, struct · line 152

Quoted 13 lines

drivers/block/nbd.c +0 -2

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

@@ -145,11 +149,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;
-

Unrelated.

Quoted 7 lines

drivers/block/nbd.c +0 -0

      nbd->blksize = blocksize;
      nbd->bytesize = (loff_t)blocksize * (loff_t)nr_blocks;
 

@@ -197,6 +199,9 @@ static void nbd_xmit_timeout(unsigned long arg)
 {
      struct nbd_device *nbd = (struct nbd_device *)arg;
 

drivers/block/nbd.c static void nbd_xmit_timeout(unsigned long arg) · line 204

Quoted 3 lines

drivers/block/nbd.c +3 -0

+     if (nbd->timedout)
+             return;
+

What does this have to do with the patch?

Quoted 14 lines

drivers/block/nbd.c +0 -2

      if (list_empty(&nbd->queue_head))
              return;
 

@@ -472,8 +477,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;

drivers/block/nbd.c static int nbd_set_socket(struct nbd_device *nbd, · line 658

Quoted 7 lines

drivers/block/nbd.c +2 -1

@@ -650,12 +653,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;

nbd->timedout is already checked in __nbd_ioctl(), no need to check it twice.

drivers/block/nbd.c static int nbd_set_socket(struct nbd_device *nbd, struct socket *sock) · line 662

Quoted 5 lines

drivers/block/nbd.c +1 -1

      else
              nbd->sock = sock;
-     spin_unlock(&nbd->sock_lock);
 
+     spin_unlock(&nbd->sock_lock);

random modification.

Quoted 30 lines

drivers/block/nbd.c +4 -1

      return ret;
 }
 

@@ -670,6 +674,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);
 }

@@ -704,6 +709,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;

@@ -733,7 +741,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: {

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

Quoted 5 lines

drivers/block/nbd.c +0 -1

@@ -752,7 +759,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);
-

random modification.

Quoted 31 lines

drivers/block/nbd.c +8 -8

              return nbd_size_set(nbd, bdev, arg, bsize);
      }

@@ -804,22 +810,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);
 

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

Quoted 3 lines

drivers/block/nbd.c +3 -0

+             mutex_lock(&bdev->bd_mutex);
+             if (!kref_get_unless_zero(&nbd->users))
+                     kref_init(&nbd->users);

This kref usage simply looks wrong and confusing. I commented last time already
that I think atomics will work better. Please discuss with me what you think before sending out a new version. Otherwise this patch series will increase in version forever.

Quoted 11 lines

drivers/block/nbd.c +4 -0

+             mutex_unlock(&bdev->bd_mutex);
+
+             mutex_lock(&nbd->tx_lock);
+             nbd_reset(nbd);
              return error;
      }
 

@@ -857,19 +870,74 @@ static int nbd_ioctl(struct block_device *bdev,
 fmode_t mode,
 

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

Quoted 8 lines

drivers/block/nbd.c +6 -0

      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);

Do we need to schedule work here?

Quoted 28 lines

drivers/block/nbd.c +27 -0

+}
+
+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)
+             );
+}

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

Quoted 10 lines

drivers/block/nbd.c +3 -0

 
 static const struct block_device_operations nbd_fops = {
      .owner =        THIS_MODULE,
      .ioctl =        nbd_ioctl,
      .compat_ioctl = nbd_ioctl,
+     .open =         nbd_open,
+     .release =      nbd_release
 };
 
+

random modification

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

Quoted 8 lines

drivers/block/nbd.c +2 -3

 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
+                                             );

...???

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

Quoted 7 lines

drivers/block/nbd.c +6 -0

+
+     struct block_device *bdev = bdget(part_devt(
+                                             dev_to_part(nbd_to_dev(nbd_dev))
+                                             )
+                                     );
+     BUG_ON(!bdev);

A simple check would be enough. Or a warning.

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

Quoted 3 lines

drivers/block/nbd.c +2 -0

+     if (nbd_dev->timedout)
+             sock_shutdown(nbd_dev);

This timeout check seems unnecessary. If we do not timeout and the socket was already closed, the sock_shutdown() will do nothing.

So if I understand you correctly you are trying to block all ioctls while you are shutting down which is a well a behaviour change of the ioctl interface. Why do you think it is better not to allow any changes until everyone closed the blockdevice? Shouldn't there be some control left for the user, for example
CLEAR_SOCK?

Regards,

Markus

Quoted 15 lines

drivers/block/nbd.c +12 -0

+
+     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)

Quoted 40 lines · Markus Pargmann

Quoted 39 lines · On 2016 M06 30, Thu 14:02:03 CEST Pranay Kr. Srivastava wrote:

When a timeout occurs or a recv fails, then instead of abruplty killing nbd block device wait for its 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 reset happens when all tasks having this bdev open closes this bdev.

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

---
 drivers/block/nbd.c | 106

++++++++++++++++++++++++++++++++++++++++++---------- 1 file changed, 87 insertions(+), 19 deletions(-)

drivers/block/nbd.c +2 -0

index e362d44..fb56dd2 100644
@@ -72,6 +72,8 @@ struct nbd_device {
 #endif
      /* 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)

@@ -99,6 +101,8 @@ static int max_part;
 static DEFINE_SPINLOCK(nbd_lock);
 

Quoted 5 lines

drivers/block/nbd.c static int max_part; · line 105

Quoted 3 lines

drivers/block/nbd.c +2 -0

 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 *);

More function signatures. Why?

To avoid code move. But do let me know why is code signature(s) like this are bad , just asking to avoid such things.

Quoted 23 lines

drivers/block/nbd.c static int nbd_size_set(struct nbd_device *nbd, struct · line 152

Quoted 13 lines

drivers/block/nbd.c +0 -2

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

@@ -145,11 +149,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;
-

Unrelated.

Quoted 7 lines

drivers/block/nbd.c +0 -0

      nbd->blksize = blocksize;
      nbd->bytesize = (loff_t)blocksize * (loff_t)nr_blocks;
 

@@ -197,6 +199,9 @@ static void nbd_xmit_timeout(unsigned long arg)
 {
      struct nbd_device *nbd = (struct nbd_device *)arg;
 

Quoted 5 lines

drivers/block/nbd.c static void nbd_xmit_timeout(unsigned long arg) · line 204

Quoted 3 lines

drivers/block/nbd.c +3 -0

+     if (nbd->timedout)
+             return;
+

What does this have to do with the patch?

to avoid re-scheduling the work function. Apparently that did cause some trouble with ext4 and 10K dd processes.

Quoted 102 lines

Quoted 15 lines

drivers/block/nbd.c +0 -2

      if (list_empty(&nbd->queue_head))
              return;
 

@@ -472,8 +477,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;

drivers/block/nbd.c static int nbd_set_socket(struct nbd_device *nbd, · line 658

Quoted 7 lines

drivers/block/nbd.c +2 -1

@@ -650,12 +653,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;

nbd->timedout is already checked in __nbd_ioctl(), no need to check it twice.

drivers/block/nbd.c static int nbd_set_socket(struct nbd_device *nbd, struct socket *sock) · line 662

Quoted 5 lines

drivers/block/nbd.c +1 -1

      else
              nbd->sock = sock;
-     spin_unlock(&nbd->sock_lock);
 
+     spin_unlock(&nbd->sock_lock);

random modification.

Quoted 30 lines

drivers/block/nbd.c +4 -1

      return ret;
 }
 

@@ -670,6 +674,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);
 }

@@ -704,6 +709,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;

@@ -733,7 +741,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: {

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

Quoted 5 lines

drivers/block/nbd.c +0 -1

@@ -752,7 +759,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);
-

random modification.

Quoted 31 lines

drivers/block/nbd.c +8 -8

              return nbd_size_set(nbd, bdev, arg, bsize);
      }

@@ -804,22 +810,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);
 

Quoted 9 lines

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

Quoted 3 lines

drivers/block/nbd.c +3 -0

+             mutex_lock(&bdev->bd_mutex);
+             if (!kref_get_unless_zero(&nbd->users))
+                     kref_init(&nbd->users);

This kref usage simply looks wrong and confusing. I commented last time already
that I think atomics will work better. Please discuss with me what you think before sending out a new version. Otherwise this patch series will increase in version forever.

Alright let's go with atomics.
But why this looks wrong, are you referring to partitioned device?

Quoted 12 lines

Quoted 12 lines

drivers/block/nbd.c +4 -0

+             mutex_unlock(&bdev->bd_mutex);
+
+             mutex_lock(&nbd->tx_lock);
+             nbd_reset(nbd);
              return error;
      }
 

@@ -857,19 +870,74 @@ static int nbd_ioctl(struct block_device *bdev,
 fmode_t mode,
 

Quoted 10 lines

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

Quoted 8 lines

drivers/block/nbd.c +6 -0

      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);

Do we need to schedule work here?

Yes this is for the kill_bdev part. This is the final kick to bdev which happens after the wait in NBD_DO_IT.

Quoted 51 lines

Quoted 29 lines

drivers/block/nbd.c +27 -0

+}
+
+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)
+             );
+}

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

Quoted 10 lines

drivers/block/nbd.c +3 -0

 
 static const struct block_device_operations nbd_fops = {
      .owner =        THIS_MODULE,
      .ioctl =        nbd_ioctl,
      .compat_ioctl = nbd_ioctl,
+     .open =         nbd_open,
+     .release =      nbd_release
 };
 
+

random modification

Quoted 8 lines

drivers/block/nbd.c +2 -3

 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
+                                             );

Quoted 1 line

...???

Tried to match the brackets... that's what you meant earlier?

Quoted 9 lines

Quoted 8 lines

drivers/block/nbd.c +6 -0

+
+     struct block_device *bdev = bdget(part_devt(
+                                             dev_to_part(nbd_to_dev(nbd_dev))
+                                             )
+                                     );
+     BUG_ON(!bdev);

Quoted 1 line

A simple check would be enough. Or a warning.

Ok, but that's really a bug.

Quoted 9 lines

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

Quoted 4 lines

drivers/block/nbd.c +2 -0

+     if (nbd_dev->timedout)
+             sock_shutdown(nbd_dev);

This timeout check seems unnecessary. If we do not timeout and the socket was already closed, the sock_shutdown() will do nothing.

Quoted 6 lines

So if I understand you correctly you are trying to block all ioctls while you are shutting down which is a well a behaviour change of the ioctl interface. Why do you think it is better not to allow any changes until everyone closed the blockdevice? Shouldn't there be some control left for the user, for example
CLEAR_SOCK?

Ah... Yes that's indeed what I'm trying to do. Now say if this block device is mounted
and another nbd-client is trying to disconnect it [CLEAR + DISCONNECT] then clear
is doing a kill_bdev. Socket already has been disconnected but the device is just not
usable in this case.

If however we are trying to provide for an error recovery, like live mounted device
and there's was timeout with all connections teared down and then someone does a set socket on this? Is this supported currently ?

A change in the CLEAR, like not actually killing bdev would also not be good. So better avoid such ioctl if device is in use, no?

Quoted 79 lines · On Sunday 10 July 2016 21:32:07 Pranay Srivastava wrote:

Quoted 40 lines · Markus Pargmann

Quoted 39 lines · On 2016 M06 30, Thu 14:02:03 CEST Pranay Kr. Srivastava wrote:

When a timeout occurs or a recv fails, then instead of abruplty killing nbd block device wait for its 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 reset happens when all tasks having this bdev open closes this bdev.

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

---
 drivers/block/nbd.c | 106

++++++++++++++++++++++++++++++++++++++++++---------- 1 file changed, 87 insertions(+), 19 deletions(-)

drivers/block/nbd.c +2 -0

index e362d44..fb56dd2 100644
@@ -72,6 +72,8 @@ struct nbd_device {
 #endif
      /* 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)

@@ -99,6 +101,8 @@ static int max_part;
 static DEFINE_SPINLOCK(nbd_lock);
 

Quoted 5 lines

drivers/block/nbd.c static int max_part; · line 105

Quoted 3 lines

drivers/block/nbd.c +2 -0

 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 *);

More function signatures. Why?

To avoid code move. But do let me know why is code signature(s) like this are bad , just asking to avoid such things.

Quoted 28 lines

drivers/block/nbd.c static int nbd_size_set(struct nbd_device *nbd, struct · line 152

Quoted 13 lines

drivers/block/nbd.c +0 -2

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

@@ -145,11 +149,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;
-

Unrelated.

drivers/block/nbd.c static void nbd_xmit_timeout(unsigned long arg) · line 204

Quoted 10 lines

drivers/block/nbd.c +3 -0

      nbd->blksize = blocksize;
      nbd->bytesize = (loff_t)blocksize * (loff_t)nr_blocks;
 

@@ -197,6 +199,9 @@ static void nbd_xmit_timeout(unsigned long arg)
 {
      struct nbd_device *nbd = (struct nbd_device *)arg;
 
+     if (nbd->timedout)
+             return;
+

What does this have to do with the patch?

Quoted 2 lines

to avoid re-scheduling the work function. Apparently that did cause some trouble with ext4 and 10K dd processes.

Ah interesting. What was the timeout in this scenario?

Quoted 113 lines

Quoted 103 lines

Quoted 16 lines

drivers/block/nbd.c +0 -2

      if (list_empty(&nbd->queue_head))
              return;
 

@@ -472,8 +477,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;

drivers/block/nbd.c static int nbd_set_socket(struct nbd_device *nbd, · line 658

Quoted 7 lines

drivers/block/nbd.c +2 -1

@@ -650,12 +653,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;

nbd->timedout is already checked in __nbd_ioctl(), no need to check it twice.

drivers/block/nbd.c static int nbd_set_socket(struct nbd_device *nbd, struct socket *sock) · line 662

Quoted 5 lines

drivers/block/nbd.c +1 -1

      else
              nbd->sock = sock;
-     spin_unlock(&nbd->sock_lock);
 
+     spin_unlock(&nbd->sock_lock);

random modification.

Quoted 30 lines

drivers/block/nbd.c +4 -1

      return ret;
 }
 

@@ -670,6 +674,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);
 }

@@ -704,6 +709,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;

@@ -733,7 +741,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: {

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

Quoted 5 lines

drivers/block/nbd.c +0 -1

@@ -752,7 +759,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);
-

random modification.

Quoted 31 lines

drivers/block/nbd.c +8 -8

              return nbd_size_set(nbd, bdev, arg, bsize);
      }

@@ -804,22 +810,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);
 

Quoted 9 lines

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

Quoted 3 lines

drivers/block/nbd.c +3 -0

+             mutex_lock(&bdev->bd_mutex);
+             if (!kref_get_unless_zero(&nbd->users))
+                     kref_init(&nbd->users);

This kref usage simply looks wrong and confusing. I commented last time already
that I think atomics will work better. Please discuss with me what you think before sending out a new version. Otherwise this patch series will increase in version forever.

Quoted 2 lines

Alright let's go with atomics.
But why this looks wrong, are you referring to partitioned device?

No, it looks wrong in respect to what kref was designed for. I really thought at the beginning that kref would work great for this setup as we have normal users that request this resource and put it back at some time (using close). But it didn't turn out so well because of this ioctl thread that keeps the file descriptor open.

So the code probably does work but the normal kref workflow with kref_init() and kref_put() simply doesn't work here.

Quoted 23 lines

Quoted 22 lines

Quoted 10 lines

drivers/block/nbd.c +4 -0

+             mutex_unlock(&bdev->bd_mutex);
+
+             mutex_lock(&nbd->tx_lock);
+             nbd_reset(nbd);
              return error;
      }
 

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

Quoted 10 lines

drivers/block/nbd.c +6 -0

@@ -857,19 +870,74 @@ 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);

Do we need to schedule work here?

Quoted 2 lines

Yes this is for the kill_bdev part. This is the final kick to bdev which happens after the wait in NBD_DO_IT.

Sorry what I meant was, whether we can directly call the appropriate function here. Without using schedule_work here. Is that possible? Or are we in some context that does not allow that?

Quoted 15 lines

Quoted 15 lines

Quoted 15 lines

drivers/block/nbd.c +12 -0

+}
+
+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)

Quoted 8 lines

Quoted 8 lines

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

Quoted 8 lines

drivers/block/nbd.c +8 -0

+             );
+     return 0;
+}
+
+static void nbd_release(struct gendisk *disk, fmode_t mode)
+{
+     struct nbd_device *nbd_dev = disk->private_data;
+

Quoted 29 lines

Quoted 29 lines

Quoted 7 lines

drivers/block/nbd.c +7 -0

+     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)
+             );
+}

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

Quoted 10 lines

drivers/block/nbd.c +3 -0

 
 static const struct block_device_operations nbd_fops = {
      .owner =        THIS_MODULE,
      .ioctl =        nbd_ioctl,
      .compat_ioctl = nbd_ioctl,
+     .open =         nbd_open,
+     .release =      nbd_release
 };
 
+

random modification

Quoted 8 lines

drivers/block/nbd.c +2 -3

 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
+                                             );

Quoted 3 lines

Quoted 1 line

...???

Tried to match the brackets... that's what you meant earlier?

Sorry seems I was unclear about that. This is what I meant:

struct nbd_device *nbd_dev = container_of(ws_nbd, struct nbd_device,
ws_shutdown);

After the line break the line should start at the beginning of the opening bracket. But closing brackets do not have to be in a separate line.

Quoted 10 lines

Quoted 10 lines

Quoted 9 lines

drivers/block/nbd.c +6 -0

+
+     struct block_device *bdev = bdget(part_devt(
+                                             dev_to_part(nbd_to_dev(nbd_dev))
+                                             )
+                                     );
+     BUG_ON(!bdev);

Quoted 3 lines

Quoted 1 line

A simple check would be enough. Or a warning.

Ok, but that's really a bug.

Yes but BUG_ON will kill the process which in this case is a worker. I think there is no need to influence anything else in the kernel as this is a nbd issue.

Quoted 25 lines

Quoted 10 lines

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

Quoted 5 lines

drivers/block/nbd.c +2 -0

+     if (nbd_dev->timedout)
+             sock_shutdown(nbd_dev);

This timeout check seems unnecessary. If we do not timeout and the socket was already closed, the sock_shutdown() will do nothing.

Quoted 6 lines

So if I understand you correctly you are trying to block all ioctls while you are shutting down which is a well a behaviour change of the ioctl interface. Why do you think it is better not to allow any changes until everyone closed the blockdevice? Shouldn't there be some control left for the user, for example
CLEAR_SOCK?

Ah... Yes that's indeed what I'm trying to do. Now say if this block device is mounted
and another nbd-client is trying to disconnect it [CLEAR + DISCONNECT] then clear
is doing a kill_bdev. Socket already has been disconnected but the device is just not
usable in this case.

Quoted 4 lines

If however we are trying to provide for an error recovery, like live mounted device
and there's was timeout with all connections teared down and then someone does a set socket on this? Is this supported currently ?

This is currently not supported. But the client has implemented something like this. So if we change this here, we should consider allowing nbd-client to react on a timeout, for example by setting a new socket.

Quoted 3 lines

A change in the CLEAR, like not actually killing bdev would also not be good. So better avoid such ioctl if device is in use, no?

What we currently have are nbd-client users that expect the device to be usable immediately after 'nbd-client -d'. Using this patch as you proposed would change this behaviour.

As an idea to fix the bug that we currently have (filesystems on blockdevice that is killed):

We could implement the killing in CLEAR_SOCK. CLEAR_SOCK is kind of a direct statement that the current socket and connection should be removed.

nbd-client currently calls CLEAR_SOCK after NBD_DO_IT, so from the users perspective with an old nbd-client, nothing changes. 'nbd-client -d' disconnects the client and leaves the blockdevice open. The following CLEAR_SOCK will kill the block device and the user does not notice a difference.

A newer nbd-client implementation could then use this new feature properly and not use CLEAR_SOCK anymore and offer something like 'nbd-client -d --force' instead. This would give the user still the possibility to have the old behaviour. But the new behaviour (keeping the blockdevice open) is the default.

Another possibility is to replace NBD_DO_IT with a new ioctl that does things differently.

Best Regards,

Markus

--

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 |

When a timeout occurs or a recv fails, then instead of abruptly killing nbd block device wait for its 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 is now refcounted with the device being released for re-use only when there are no "other users".

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

Behavioral Change:

1) NBD_DO_IT will not wait for the device to be reset. Hence
the nbd-client "may exit" while some other process is using this device without actually doing the reset on this device , hence thus making it unusable until all such user space processes have stopped using this device.

2) There's a window where the nbd-client will not be able to
change / issue ioctls to the nbd device. This is when there's been a disconnect issued or a timeout has occured, however there are "other" user processes which currently have this nbd device opened.

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

---
 drivers/block/nbd.c | 37 ++++++++++++++++++++++++++++++++-----
 1 file changed, 32 insertions(+), 5 deletions(-)

drivers/block/nbd.c +32 -6

index 4919760..fe36280 100644
@@ -74,6 +74,7 @@ struct nbd_device {
         *This is specifically for calling sock_shutdown, for now.
         */
        struct work_struct ws_shutdown;
+       atomic_t users; /* Users that opened the block device */
 };
 
 #if IS_ENABLED(CONFIG_DEBUG_FS)

@@ -699,6 +700,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->disconnect || nbd->timedout)
+               return -EBUSY;
+
        switch (cmd) {
        case NBD_DISCONNECT: {
                struct request sreq;

@@ -728,7 +732,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: {

@@ -804,16 +807,12 @@ static int __nbd_ioctl(struct block_device *bdev, struct nbd_device *nbd,
                mutex_lock(&nbd->tx_lock);
                nbd->task_recv = NULL;
                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;
 
-               nbd_reset(nbd);
-
                return error;
        }
 

@@ -852,10 +851,38 @@ static int nbd_ioctl(struct block_device *bdev, fmode_t mode,
        return error;
 }
 
+static int nbd_open(struct block_device *bdev, fmode_t mode)
+{
+       struct nbd_device *nbd = bdev->bd_disk->private_data;
+
+       atomic_inc(&nbd->users);
+
+       return 0;
+}
+
+static void nbd_release(struct gendisk *disk, fmode_t mode)
+{
+       struct nbd_device *nbd = disk->private_data;
+       struct block_device *bdev = bdget (part_devt(
+                                               dev_to_part(nbd_to_dev(nbd))));
+
+       WARN_ON(!bdev);
+       if (atomic_dec_and_test(&nbd->users)) {
+               if (bdev) {
+                       nbd_bdev_reset(bdev);
+                       kill_bdev(bdev);
+                       bdput(bdev);
+               }
+               nbd_reset(nbd);
+       }
+}
+
 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)
--

2.6.2

When a timeout occurs or a recv fails, then instead of abruptly killing nbd block device wait for its 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 is now refcounted with the device being released for re-use only when there are no "other users".

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

Behavioral Change:

1) NBD_DO_IT will not wait for the device to be reset. Hence
the nbd-client "may exit" while some other process is using this device without actually doing the reset on this device , hence thus making it unusable until all such user space processes have stopped using this device.

2) There's a window where the nbd-client will not be able to
change / issue ioctls to the nbd device. This is when there's been a disconnect issued or a timeout has occured, however there are "other" user processes which currently have this nbd device opened.

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

---
 drivers/block/nbd.c | 37 ++++++++++++++++++++++++++++++++-----
 1 file changed, 32 insertions(+), 5 deletions(-)

drivers/block/nbd.c +32 -5

index 4919760..fe36280 100644
@@ -74,6 +74,7 @@ struct nbd_device {
 	 *This is specifically for calling sock_shutdown, for now.
 	 */
 	struct work_struct ws_shutdown;
+	atomic_t users; /* Users that opened the block device */
 };
 
 #if IS_ENABLED(CONFIG_DEBUG_FS)

@@ -699,6 +700,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->disconnect || nbd->timedout)
+		return -EBUSY;
+
 	switch (cmd) {
 	case NBD_DISCONNECT: {
 		struct request sreq;

@@ -728,7 +732,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: {

@@ -804,16 +807,12 @@ static int __nbd_ioctl(struct block_device *bdev, struct nbd_device *nbd,
 		mutex_lock(&nbd->tx_lock);
 		nbd->task_recv = NULL;
 		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;
 
-		nbd_reset(nbd);
-
 		return error;
 	}
 

@@ -852,10 +851,38 @@ static int nbd_ioctl(struct block_device *bdev, fmode_t mode,
 	return error;
 }
 
+static int nbd_open(struct block_device *bdev, fmode_t mode)
+{
+	struct nbd_device *nbd = bdev->bd_disk->private_data;
+
+	atomic_inc(&nbd->users);
+
+	return 0;
+}
+
+static void nbd_release(struct gendisk *disk, fmode_t mode)
+{
+	struct nbd_device *nbd = disk->private_data;
+	struct block_device *bdev = bdget (part_devt(
+						dev_to_part(nbd_to_dev(nbd))));
+
+	WARN_ON(!bdev);
+	if (atomic_dec_and_test(&nbd->users)) {
+		if (bdev) {
+			nbd_bdev_reset(bdev);
+			kill_bdev(bdev);
+			bdput(bdev);
+		}
+		nbd_reset(nbd);
+	}
+}
+
 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)