[PATCH v3 15/15] rbd: add support for feature of RBD_FEATURE_JOURNALING

[Dongsheng Yang <dongsheng.yang@xxxxxxxxxxxx>]

Allow user to map rbd images with journaling enabled, but
there is a warning in demsg:

        WARNING: kernel journaling is EXPERIMENTAL!

Signed-off-by: Dongsheng Yang <dongsheng.yang@xxxxxxxxxxxx>
---
 drivers/block/rbd.c | 99 +++++++++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 99 insertions(+)

diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
index 3c44db7..3e940a7 100644
--- a/drivers/block/rbd.c
+++ b/drivers/block/rbd.c
@@ -131,6 +131,7 @@ static int atomic_dec_return_safe(atomic_t *v)
 				 RBD_FEATURE_OBJECT_MAP |	\
 				 RBD_FEATURE_FAST_DIFF |	\
 				 RBD_FEATURE_DEEP_FLATTEN |	\
+				 RBD_FEATURE_JOURNALING |	\
 				 RBD_FEATURE_DATA_POOL |	\
 				 RBD_FEATURE_OPERATIONS)
 
@@ -3847,6 +3848,7 @@ static void __rbd_lock(struct rbd_device *rbd_dev, const char *cookie)
 /*
  * lock_rwsem must be held for write
  */
+static int rbd_dev_open_journal(struct rbd_device *rbd_dev);
 static int rbd_lock(struct rbd_device *rbd_dev)
 {
 	struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
@@ -4190,6 +4192,12 @@ static int rbd_post_acquire_action(struct rbd_device *rbd_dev)
 			return ret;
 	}
 
+	if (rbd_dev->header.features & RBD_FEATURE_JOURNALING) {
+		ret = rbd_dev_open_journal(rbd_dev);
+		if (ret)
+			return ret;
+	}
+
 	return 0;
 }
 
@@ -4320,10 +4328,14 @@ static bool rbd_quiesce_lock(struct rbd_device *rbd_dev)
 	return true;
 }
 
+static void rbd_dev_close_journal(struct rbd_device *rbd_dev);
 static void rbd_pre_release_action(struct rbd_device *rbd_dev)
 {
 	if (rbd_dev->header.features & RBD_FEATURE_OBJECT_MAP)
 		rbd_object_map_close(rbd_dev);
+
+	if (rbd_dev->header.features & RBD_FEATURE_JOURNALING)
+		rbd_dev_close_journal(rbd_dev);
 }
 
 static void __rbd_release_lock(struct rbd_device *rbd_dev)
@@ -7303,6 +7315,88 @@ static int rbd_journal_replay(void *entry_handler, struct ceph_journaler_entry *
 	return ret;
 }
 
+static int rbd_journal_allocate_tag(struct rbd_journal *journal);
+static int rbd_journal_open(struct rbd_journal *journal)
+{
+	struct ceph_journaler *journaler = journal->journaler;
+	int ret;
+
+	ret = ceph_journaler_open(journaler);
+	if (ret)
+		goto out;
+
+	ret = ceph_journaler_start_replay(journaler);
+	if (ret)
+		goto err_close_journaler;
+
+	ret = rbd_journal_allocate_tag(journal);
+	if (ret)
+		goto err_close_journaler;
+	return ret;
+
+err_close_journaler:
+	ceph_journaler_close(journaler);
+
+out:
+	return ret;
+}
+
+static int rbd_dev_open_journal(struct rbd_device *rbd_dev)
+{
+	int ret;
+	struct rbd_journal *journal;
+	struct ceph_journaler *journaler;
+	struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
+
+	// create journal
+	rbd_assert(!rbd_dev->journal);
+
+	journal = kzalloc(sizeof(struct rbd_journal), GFP_KERNEL);
+	if (!journal)
+		return -ENOMEM;
+
+	journaler = ceph_journaler_create(osdc, &rbd_dev->header_oloc,
+					  rbd_dev->spec->image_id,
+					  LOCAL_CLIENT_ID);
+	if (!journaler) {
+		ret = -ENOMEM;
+		goto err_free_journal;
+	}
+
+	// journal init
+	journaler->entry_handler = rbd_dev;
+	journaler->handle_entry = rbd_journal_replay;
+
+	journal->journaler = journaler;
+	rbd_dev->journal = journal;
+
+	// journal open
+	ret = rbd_journal_open(rbd_dev->journal);
+	if (ret)
+		goto err_destroy_journaler;
+
+	return ret;
+
+err_destroy_journaler:
+	ceph_journaler_destroy(journaler);
+err_free_journal:
+	kfree(rbd_dev->journal);
+	rbd_dev->journal = NULL;
+	return ret;
+}
+
+static void rbd_dev_close_journal(struct rbd_device *rbd_dev)
+{
+	rbd_assert(rbd_dev->journal);
+
+	ceph_journaler_close(rbd_dev->journal->journaler);
+	rbd_dev->journal->tag_tid = 0;
+
+	ceph_journaler_destroy(rbd_dev->journal->journaler);
+	kfree(rbd_dev->journal);
+	rbd_dev->journal = NULL;
+}
+
 typedef struct rbd_journal_tag_predecessor {
 	bool commit_valid;
 	uint64_t tag_tid;
@@ -7493,6 +7587,11 @@ static int rbd_dev_image_probe(struct rbd_device *rbd_dev, int depth)
 			goto err_out_probe;
 	}
 
+	if (rbd_dev->header.features & RBD_FEATURE_JOURNALING) {
+		rbd_warn(rbd_dev,
+			 "WARNING: kernel rbd journaling is EXPERIMENTAL!");
+	}
+
 	ret = rbd_dev_probe_parent(rbd_dev, depth);
 	if (ret)
 		goto err_out_probe;
-- 
1.8.3.1