ocf: implement OCF cleaner

Implement cleaner that is used in WriteBack mode.

Cleaner is a background agent that does synchronization
 of data between cache and its cores.
 Cleaner usually runs every ~20 seconds to perform cleaning.
 The synchronization is a simmilar operation to OCF management flushes.
We need cleaner for WriteBack because only WriteBack mode
 produces dirty data that cleaner needs to deal with.
Cleaner requires adopting trylock() because in current version
 cleaner uses management lock when performs cleaning,
 which may lead to deadlocks if cleaner runs on
 the same thread as management operations.

WriteBack mode is fully functional after this change,
  but persistent metadata support is required to use it for production.

Cleaner will run on management thread for now.
  We plan to implement functionality of
  chosing a CPU core where cleaner should run.
Cleaning policy is not configurable yet.
  The default is ALRU with 20 sec interval.

Signed-off-by: Vitaliy Mysak <vitaliy.mysak@intel.com>
Reviewed-on: https://review.gerrithub.io/c/spdk/spdk/+/448537 (master)

(cherry picked from commit 80a2ff01f3)
Change-Id: I35aa7e00c44e0d7a77e64e60df1f66f20be03f55
Signed-off-by: Tomasz Zawadzki <tomasz.zawadzki@intel.com>
Reviewed-on: https://review.gerrithub.io/c/spdk/spdk/+/457258
Reviewed-by: Darek Stojaczyk <dariusz.stojaczyk@intel.com>
Reviewed-by: Ben Walker <benjamin.walker@intel.com>
Tested-by: SPDK CI Jenkins <sys_sgci@intel.com>

lib/bdev/ocf/ctx.c

@@ -322,17 +322,109 @@ void vbdev_ocf_cache_ctx_get(struct vbdev_ocf_cache_ctx *ctx)
 	env_atomic_inc(&ctx->refcnt);
 }
 
+struct cleaner_priv {
+	struct spdk_poller *poller;
+	ocf_queue_t         queue;
+	uint64_t            next_run;
+};
+
+static int
+cleaner_poll(void *arg)
+{
+	ocf_cleaner_t cleaner = arg;
+	struct cleaner_priv *priv = ocf_cleaner_get_priv(cleaner);
+	uint32_t iono = ocf_queue_pending_io(priv->queue);
+	int i, max = spdk_min(32, iono);
+
+	for (i = 0; i < max; i++) {
+		ocf_queue_run_single(priv->queue);
+	}
+
+	if (spdk_get_ticks() >= priv->next_run) {
+		ocf_cleaner_run(cleaner, priv->queue);
+		return 1;
+	}
+
+	if (iono > 0) {
+		return 1;
+	} else {
+		return 0;
+	}
+}
+
+static void
+cleaner_cmpl(ocf_cleaner_t c, uint32_t interval)
+{
+	struct cleaner_priv *priv = ocf_cleaner_get_priv(c);
+
+	priv->next_run = spdk_get_ticks() + ((interval * spdk_get_ticks_hz()) / 1000);
+}
+
+static void
+cleaner_queue_kick(ocf_queue_t q)
+{
+}
+
+static void
+cleaner_queue_stop(ocf_queue_t q)
+{
+	struct cleaner_priv *cpriv = ocf_queue_get_priv(q);
+
+	if (cpriv) {
+		spdk_poller_unregister(&cpriv->poller);
+		free(cpriv);
+	}
+}
+
+const struct ocf_queue_ops cleaner_queue_ops = {
+	.kick_sync = cleaner_queue_kick,
+	.kick = cleaner_queue_kick,
+	.stop = cleaner_queue_stop,
+};
+
 static int
 vbdev_ocf_ctx_cleaner_init(ocf_cleaner_t c)
 {
-	/* TODO [writeback]: implement with writeback mode support */
+	int rc;
+	struct cleaner_priv        *priv  = calloc(1, sizeof(*priv));
+	ocf_cache_t                 cache = ocf_cleaner_get_cache(c);
+	struct vbdev_ocf_cache_ctx *cctx  = ocf_cache_get_priv(cache);
+
+	if (priv == NULL) {
+		return -ENOMEM;
+	}
+
+	rc = vbdev_ocf_queue_create(cache, &priv->queue, &cleaner_queue_ops);
+	if (rc) {
+		free(priv);
+		return rc;
+	}
+
+	/* We start cleaner poller at the same thread where cache was created
+	 * TODO: allow user to specify core at which cleaner should run */
+	priv->poller = spdk_poller_register(cleaner_poll, c, 0);
+	if (priv->poller == NULL) {
+		vbdev_ocf_queue_put(priv->queue);
+		free(priv);
+		return -ENOMEM;
+	}
+
+	ocf_queue_set_priv(priv->queue, priv);
+
+	cctx->cleaner_queue  = priv->queue;
+
+	ocf_cleaner_set_cmpl(c, cleaner_cmpl);
+	ocf_cleaner_set_priv(c, priv);
+
 	return 0;
 }
 
 static void
 vbdev_ocf_ctx_cleaner_stop(ocf_cleaner_t c)
 {
-	/* TODO [writeback]: implement with writeback mode support */
+	struct cleaner_priv *priv = ocf_cleaner_get_priv(c);
+
+	vbdev_ocf_queue_put(priv->queue);
 }
 
 static int vbdev_ocf_volume_updater_init(ocf_metadata_updater_t mu)

lib/bdev/ocf/ctx.h

@@ -46,6 +46,7 @@ extern ocf_ctx_t vbdev_ocf_ctx;
 /* Context of cache instance */
 struct vbdev_ocf_cache_ctx {
 	ocf_queue_t                  mngt_queue;
+	ocf_queue_t                  cleaner_queue;
 	struct spdk_io_channel      *management_channel;
 	pthread_mutex_t              lock;
 	env_atomic                   refcnt;

lib/bdev/ocf/volume.c

@@ -167,7 +167,6 @@ vbdev_ocf_volume_submit_io_cb(struct spdk_bdev_io *bdev_io, bool success, void *
 
 	io = opaque;
 	io_ctx = ocf_get_io_ctx(io);
-
 	assert(io_ctx != NULL);
 
 	if (!success) {
@@ -180,6 +179,9 @@ vbdev_ocf_volume_submit_io_cb(struct spdk_bdev_io *bdev_io, bool success, void *
 		case SPDK_BDEV_IO_TYPE_WRITE:
 			env_free(bdev_io->u.bdev.iovs);
 			break;
+		case SPDK_BDEV_IO_TYPE_FLUSH:
+			/* No iovs were allocated for flush request */
+			break;
 		default:
 			assert(false);
 			break;
@@ -237,7 +239,7 @@ prepare_submit(struct ocf_io *io)
 	cache = ocf_queue_get_cache(q);
 	cctx = ocf_cache_get_priv(cache);
 
-	if (q == cctx->mngt_queue) {
+	if (q == cctx->cleaner_queue || q == cctx->mngt_queue) {
 		io_ctx->ch = base->management_channel;
 		return 0;
 	}
@@ -263,11 +265,6 @@ vbdev_ocf_volume_submit_flush(struct ocf_io *io)
 	struct ocf_io_ctx *io_ctx = ocf_get_io_ctx(io);
 	int status;
 
-	if (base->is_cache) {
-		io->end(io, 0);
-		return;
-	}
-
 	status = prepare_submit(io);
 	if (status) {
 		SPDK_ERRLOG("Preparing io failed with status=%d\n", status);