[Devel] [PATCH VZ9 3/3] fs/fuse: pcs: improve workqueue scheduling

[Alexey Kuznetsov]

Add configurable flags to control workqueue scheduling and job
forwarding to workqueues.

1. enable_cpu_wq - create the second workqueue pool, which
   is intended to perform cpu intensive never sleeping works.
   Default: 1
2. worker_flags  - allow to tune flags on main workqueue.
   0 - nothing. Default
   1 - WQ_CPU_INTENSIVE
   2 - WQ_UNBOUND
3. cpu_worker_flags - like worker_flags, but for new cpu_wq.
   Default is 1 i.e. WQ_CPU_INTENSIVE
3. crc_verify - enable crc checking
   Default is 1.

Default settings are selected to solve VSTOR-83607
in the safest way. Possibility to tune the options at runtime
is added, because investigation of real mechanism is still ongoing,
it is not obvious why plain original setup is not enough, this
could be just a bug/feature in core workqueue scheduler.

https://pmc.acronis.work/browse/VSTOR-83607

Signed-off-by: Alexey Kuznetsov <kuznet at acronis.com>
---
 fs/fuse/kio/pcs/pcs_cluster.h      |  1 +
 fs/fuse/kio/pcs/pcs_cs_accel.c     | 14 +++++++++++---
 fs/fuse/kio/pcs/pcs_fuse_kdirect.c | 23 +++++++++++++++++++++--
 3 files changed, 33 insertions(+), 5 deletions(-)

diff --git a/fs/fuse/kio/pcs/pcs_cluster.h b/fs/fuse/kio/pcs/pcs_cluster.h
index 266d8e5..a69fb6f 100644
--- a/fs/fuse/kio/pcs/pcs_cluster.h
+++ b/fs/fuse/kio/pcs/pcs_cluster.h
@@ -64,6 +64,7 @@ struct pcs_fuse_work {
 };
 
 extern struct workqueue_struct *pcs_cleanup_wq;
+extern struct workqueue_struct *pcs_cpu_wq;
 
 int pcs_cluster_init(struct pcs_fuse_cluster *c, struct workqueue_struct *,
 		     struct fuse_conn *fc, struct pcs_ioc_init_kdirect *info);
diff --git a/fs/fuse/kio/pcs/pcs_cs_accel.c b/fs/fuse/kio/pcs/pcs_cs_accel.c
index 067598b..a18069f 100644
--- a/fs/fuse/kio/pcs/pcs_cs_accel.c
+++ b/fs/fuse/kio/pcs/pcs_cs_accel.c
@@ -22,6 +22,14 @@
 #include "log.h"
 #include "fuse_ktrace.h"
 
+static unsigned int crc_verify = 1;
+module_param(crc_verify, uint, 0644);
+MODULE_PARM_DESC(crc_verify, "Enable CRC check");
+
+static unsigned int enable_cpu_wq;
+module_param(enable_cpu_wq, uint, 0644);
+MODULE_PARM_DESC(enable_cpu_wq, "Execute crypto/crc in separate work pool");
+
 /* CSA context can be referenced from two places:
  *  * csaccel file struct as filp->private_data
  *    This reference is dropped at csaccel file close
@@ -519,7 +527,7 @@ static void pcs_csa_complete(struct kiocb *iocb, long ret)
 
 	if (atomic_dec_and_test(&areq->iocount)) {
 		INIT_WORK(&areq->work, csa_complete_work);
-		queue_work(ireq->cc->wq, &areq->work);
+		queue_work(enable_cpu_wq ? pcs_cpu_wq : ireq->cc->wq, &areq->work);
 	}
 }
 
@@ -608,7 +616,7 @@ static inline int csa_submit(struct file * file, struct file *cfile, int do_csum
 	 */
 	if (atomic_dec_and_test(&areq->iocount)) {
 		INIT_WORK(&areq->work, csa_complete_work);
-		queue_work(ireq->cc->wq, &areq->work);
+		queue_work(enable_cpu_wq ? pcs_cpu_wq : ireq->cc->wq, &areq->work);
 	}
 	return 0;
 }
@@ -631,7 +639,7 @@ int pcs_csa_cs_submit(struct pcs_cs * cs, struct pcs_int_request * ireq)
 			if (!(map->state & PCS_MAP_DEAD) && map->cs_list == ireq->iochunk.csl) {
 				struct file * file = get_file(csa->file);
 				struct file * cfile = csa->cfile ? get_file(csa->cfile) : NULL;
-				unsigned int flags = csa->flags;
+				unsigned int flags = crc_verify ? csa->flags : 0;
 				int err;
 
 				if (csa_ctx->tfm)
diff --git a/fs/fuse/kio/pcs/pcs_fuse_kdirect.c b/fs/fuse/kio/pcs/pcs_fuse_kdirect.c
index 4180aa8..ff47653 100644
--- a/fs/fuse/kio/pcs/pcs_fuse_kdirect.c
+++ b/fs/fuse/kio/pcs/pcs_fuse_kdirect.c
@@ -71,6 +71,16 @@
 module_param(rdmaio_queue_depth, uint, 0644);
 MODULE_PARM_DESC(rdmaio_queue_depth, "RDMA queue depth");
 
+unsigned int worker_flags;
+module_param(worker_flags, uint, 0444);
+MODULE_PARM_DESC(worker_flags, "Set main worker flags");
+
+unsigned int cpu_worker_flags = 1;
+module_param(cpu_worker_flags, uint, 0444);
+MODULE_PARM_DESC(cpu_worker_flags, "Set cpu worker flags");
+
+#define sel_wq_flags(f) ((((f) & 1) ? WQ_CPU_INTENSIVE : 0) | (((f) & 2) ? WQ_UNBOUND : 0))
+
 void (*fuse_printk_plugin)(unsigned long, const char *, ...);
 EXPORT_SYMBOL(fuse_printk_plugin);
 
@@ -145,6 +155,7 @@ static int set_io_fail_percent(const char *val, const struct kernel_param *kp)
 static struct kmem_cache *pcs_fuse_req_cachep;
 static struct kmem_cache *pcs_ireq_cachep;
 static struct workqueue_struct *pcs_wq;
+struct workqueue_struct *pcs_cpu_wq;
 struct workqueue_struct *pcs_cleanup_wq;
 static struct fuse_kio_ops kio_pcs_ops;
 static struct dentry *fuse_trace_root;
@@ -289,6 +300,7 @@ static void kpcs_conn_fini(struct fuse_mount *fm)
 	unregister_client(fc->kio.ctx);
 	synchronize_rcu();
 	flush_workqueue(pcs_wq);
+	flush_workqueue(pcs_cpu_wq);
 	flush_workqueue(pcs_cleanup_wq);
 	pcs_cluster_fini((struct pcs_fuse_cluster *) fc->kio.ctx);
 }
@@ -1909,13 +1921,17 @@ static int __init kpcs_mod_init(void)
 	if (!pcs_map_cachep)
 		goto free_ireq_cache;
 
-	pcs_wq = alloc_workqueue("pcs_cluster", WQ_MEM_RECLAIM, 0);
+	pcs_wq = alloc_workqueue("pcs_cluster", WQ_MEM_RECLAIM|sel_wq_flags(worker_flags), 0);
 	if (!pcs_wq)
 		goto free_map_cache;
 
+	pcs_cpu_wq = alloc_workqueue("pcs_cpu", WQ_MEM_RECLAIM|sel_wq_flags(cpu_worker_flags), 0);
+	if (!pcs_cpu_wq)
+		goto free_wq;
+
 	pcs_cleanup_wq = alloc_workqueue("pcs_cleanup_wq", WQ_MEM_RECLAIM, 0);
 	if (!pcs_cleanup_wq)
-		goto free_wq;
+		goto free_cpu_wq;
 
 	if (pcs_csa_init())
 		goto free_cleanup_wq;
@@ -1946,6 +1962,8 @@ static int __init kpcs_mod_init(void)
 	pcs_csa_fini();
 free_cleanup_wq:
 	destroy_workqueue(pcs_cleanup_wq);
+free_cpu_wq:
+	destroy_workqueue(pcs_cpu_wq);
 free_wq:
 	destroy_workqueue(pcs_wq);
 free_map_cache:
@@ -1966,6 +1984,7 @@ static void __exit kpcs_mod_exit(void)
 
 	fuse_unregister_kio(&kio_pcs_ops);
 	destroy_workqueue(pcs_cleanup_wq);
+	destroy_workqueue(pcs_cpu_wq);
 	destroy_workqueue(pcs_wq);
 	kmem_cache_destroy(pcs_map_cachep);
 	kmem_cache_destroy(pcs_ireq_cachep);
-- 
1.8.3.1