rbd: use watch/notify for changes in rbd header

Send notifications when we change the rbd header (e.g. create a snapshot)
and wait for such notifications.  This allows synchronizing the snapshot
creation between different rbd clients/rools.

Signed-off-by: Yehuda Sadeh <yehuda@hq.newdream.net>
Signed-off-by: Sage Weil <sage@newdream.net>

1 files changed, 335 insertions, 26 deletions

diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
index e1e38b11f48a..16dc3645291c 100644
--- a/drivers/block/rbd.c
+++ b/drivers/block/rbd.c
@@ -31,6 +31,7 @@
 #include <linux/ceph/osd_client.h>
 #include <linux/ceph/mon_client.h>
 #include <linux/ceph/decode.h>
+#include <linux/parser.h>
 
 #include <linux/kernel.h>
 #include <linux/device.h>
@@ -54,6 +55,8 @@
 
 #define DEV_NAME_LEN            32
 
+#define RBD_NOTIFY_TIMEOUT_DEFAULT 10
+
 /*
  * block device image metadata (in-memory version)
  */
@@ -71,6 +74,12 @@ struct rbd_image_header {
 
         char *snap_names;
         u64 *snap_sizes;
+
+        u64 obj_version;
+};
+
+struct rbd_options {
+        int     notify_timeout;
 };
 
 /*
@@ -78,6 +87,7 @@ struct rbd_image_header {
  */
 struct rbd_client {
         struct ceph_client      *client;
+        struct rbd_options      *rbd_opts;
         struct kref             kref;
         struct list_head        node;
 };
@@ -124,6 +134,9 @@ struct rbd_device {
         char                    pool_name[RBD_MAX_POOL_NAME_LEN];
         int                     poolid;
 
+        struct ceph_osd_event   *watch_event;
+        struct ceph_osd_request *watch_request;
+
         char                    snap_name[RBD_MAX_SNAP_NAME_LEN];
         u32 cur_snap;   /* index+1 of current snapshot within snap context
                            0 - for the head */
@@ -177,6 +190,8 @@ static void rbd_put_dev(struct rbd_device *rbd_dev)
         put_device(&rbd_dev->dev);
 }
 
+static int __rbd_update_snaps(struct rbd_device *rbd_dev);
+
 static int rbd_open(struct block_device *bdev, fmode_t mode)
 {
         struct gendisk *disk = bdev->bd_disk;
@@ -211,7 +226,8 @@ static const struct block_device_operations rbd_bd_ops = {
  * Initialize an rbd client instance.
  * We own *opt.
  */
-static struct rbd_client *rbd_client_create(struct ceph_options *opt)
+static struct rbd_client *rbd_client_create(struct ceph_options *opt,
+                                            struct rbd_options *rbd_opts)
 {
         struct rbd_client *rbdc;
         int ret = -ENOMEM;
@@ -233,6 +249,8 @@ static struct rbd_client *rbd_client_create(struct ceph_options *opt)
         if (ret < 0)
                 goto out_err;
 
+        rbdc->rbd_opts = rbd_opts;
+
         spin_lock(&node_lock);
         list_add_tail(&rbdc->node, &rbd_client_list);
         spin_unlock(&node_lock);
@@ -267,6 +285,59 @@ static struct rbd_client *__rbd_client_find(struct ceph_options *opt)
 }
 
 /*
+ * mount options
+ */
+enum {
+        Opt_notify_timeout,
+        Opt_last_int,
+        /* int args above */
+        Opt_last_string,
+        /* string args above */
+};
+
+static match_table_t rbdopt_tokens = {
+        {Opt_notify_timeout, "notify_timeout=%d"},
+        /* int args above */
+        /* string args above */
+        {-1, NULL}
+};
+
+static int parse_rbd_opts_token(char *c, void *private)
+{
+        struct rbd_options *rbdopt = private;
+        substring_t argstr[MAX_OPT_ARGS];
+        int token, intval, ret;
+
+        token = match_token((char *)c, rbdopt_tokens, argstr);
+        if (token < 0)
+                return -EINVAL;
+
+        if (token < Opt_last_int) {
+                ret = match_int(&argstr[0], &intval);
+                if (ret < 0) {
+                        pr_err("bad mount option arg (not int) "
+                               "at '%s'\n", c);
+                        return ret;
+                }
+                dout("got int token %d val %d\n", token, intval);
+        } else if (token > Opt_last_int && token < Opt_last_string) {
+                dout("got string token %d val %s\n", token,
+                     argstr[0].from);
+        } else {
+                dout("got token %d\n", token);
+        }
+
+        switch (token) {
+        case Opt_notify_timeout:
+                rbdopt->notify_timeout = intval;
+                break;
+        default:
+                BUG_ON(token);
+        }
+        return 0;
+}
+
+/*
  * Get a ceph client with specific addr and configuration, if one does
  * not exist create it.
  */
@@ -276,11 +347,18 @@ static int rbd_get_client(struct rbd_device *rbd_dev, const char *mon_addr,
         struct rbd_client *rbdc;
         struct ceph_options *opt;
         int ret;
+        struct rbd_options *rbd_opts;
+
+        rbd_opts = kzalloc(sizeof(*rbd_opts), GFP_KERNEL);
+        if (!rbd_opts)
+                return -ENOMEM;
+
+        rbd_opts->notify_timeout = RBD_NOTIFY_TIMEOUT_DEFAULT;
 
         ret = ceph_parse_options(&opt, options, mon_addr,
-                                 mon_addr + strlen(mon_addr), NULL, NULL);
+                                 mon_addr + strlen(mon_addr), parse_rbd_opts_token, rbd_opts);
         if (ret < 0)
-                return ret;
+                goto done_err;
 
         spin_lock(&node_lock);
         rbdc = __rbd_client_find(opt);
@@ -296,13 +374,18 @@ static int rbd_get_client(struct rbd_device *rbd_dev, const char *mon_addr,
         }
         spin_unlock(&node_lock);
 
-        rbdc = rbd_client_create(opt);
-        if (IS_ERR(rbdc))
-                return PTR_ERR(rbdc);
+        rbdc = rbd_client_create(opt, rbd_opts);
+        if (IS_ERR(rbdc)) {
+                ret = PTR_ERR(rbdc);
+                goto done_err;
+        }
 
         rbd_dev->rbd_client = rbdc;
         rbd_dev->client = rbdc->client;
         return 0;
+done_err:
+        kfree(rbd_opts);
+        return ret;
 }