rds: Put back pages on the CPU that allocated them

The RDBMS usage model for RDS dictates that an RDS message, which is
an SG list populated with enough order zero pages to hold the RDS
messages, is allocated where the RDBMS process runs. And they run on
any CPU.

The Reliability aspect of RDS requires RDS to keep the message until
the peer has acknowledged it. This happens by an explicit or an
implicit ACK. When said ACKs are received, the pages in the SG list
are freed (put). However, this only happens on the NUMA node local to
the HCA used for communication, due to how cellirqbalance works.

The above facts lead to a surplus of order zero pages on the NUMA node
local to the HCA, and similar, a deficit of order zero pages on other
NUMA nodes. Even though the SLAB allocation system is supposedly
lock-free for order zero pages, locks are taken in the above scenario
in order to establish balance again.

This will again lead to lock contention in the kernel, and reduced
RDBMS IOPS for certain workloads.

This is fixed by maintaining a per-cpu cache for pages. In
rds_copy_from_user(), we allocate from the per-cpu cache if
possible. Likewise, when purging RDS messages, we put the pages back
on the per-cpu list that allocated the page, unless it is too long.

To avoid pages being stuck forever in the per-cpu cache, we have a
garbage collector, which by default runs every second and cleans 10%
of the possible CPU caches per invocation.

When the RDS module is removed, we stop the worker thread and flushes
100% of all possible CPUs.

This optimization provides an 11% improvement in update IOPS on an X10M
VM running Oracle RDBMS.

Orabug: 35768362

Suggested-by: Jane Chu <[email protected]>
Signed-off-by: Håkon Bugge <[email protected]>
Tested-by: Håkon Bugge <[email protected]>
Tested-by: Shih-Yu Huang <[email protected]>
Reviewed-by: William Kucharski <[email protected]>
Tested-by: Alexis Silva <[email protected]>

Orabug: 35768362

LUCI => v6.11

Conflicts:
	net/rds/sysctl.c
	- Merge conflict due to the missing ctl_table sentinel (Orabug 36936368)

Reviewed-by: Hans Westgaard Ry <[email protected]>
Signed-off-by: Håkon Bugge <[email protected]>

Author: Hakon-Bugge

net/rds/af_rds.c

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2006, 2021 Oracle and/or its affiliates.
+ * Copyright (c) 2006, 2023, Oracle and/or its affiliates.
  *
  * This software is available to you under a choice of one of two
  * licenses.  You may choose to be licensed under the terms of the GNU
@@ -1449,6 +1449,7 @@ static void __exit rds_exit(void)
 	rds_info_deregister_func(RDS6_INFO_RECV_MESSAGES, rds6_sock_inc_info);
 #endif
 	kmem_cache_destroy(rds_rs_buf_info_slab);
+	rds_cfu_fini_cache();
 }
 
 module_exit(rds_exit);
@@ -1503,6 +1504,7 @@ static int __init rds_init(void)
 #endif
 
 	rds_qos_threshold_init();
+	rds_cfu_init_cache();
 
 	goto out;
 

net/rds/message.c

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2006, 2020 Oracle and/or its affiliates.
+ * Copyright (c) 2006, 2023, Oracle and/or its affiliates.
  *
  * This software is available to you under a choice of one of two
  * licenses.  You may choose to be licensed under the terms of the GNU
@@ -31,8 +31,10 @@
  *
  */
 #include <linux/kernel.h>
+#include <linux/topology.h>
 
 #include "rds.h"
+#include "lfstack.h"
 
 static unsigned int	rds_exthdr_size[] = {
 [RDS_EXTHDR_NONE]	= 0,
@@ -47,6 +49,22 @@ static unsigned int	rds_exthdr_size[] = {
 [RDS_EXTHDR_CSUM]       = sizeof(struct rds_ext_header_rdma_csum),
 };
 
+struct rds_cfu_cache_entry {
+	struct page *pg;
+	struct lfstack_el list;
+};
+
+struct rds_cfu_gc_control {
+	int percent_cpus_to_clean;
+	int next_cpu;
+	struct delayed_work work;
+};
+
+static struct rds_cfu_gc_control gc_control;
+
+DEFINE_PER_CPU(union lfstack, rds_cfu_cache) ____cacheline_aligned;
+DEFINE_PER_CPU(atomic_t, rds_cfu_entries) ____cacheline_aligned;
+static bool rds_cfu_cache_tearing_down;
 
 void rds_message_addref(struct rds_message *rm)
 {
@@ -55,21 +73,108 @@ void rds_message_addref(struct rds_message *rm)
 }
 EXPORT_SYMBOL_GPL(rds_message_addref);
 
+static void rds_cfu_cache_do_gc(void)
+{
+	int cpus_to_check = num_possible_cpus() * gc_control.percent_cpus_to_clean / 100;
+	int i;
+
+	if (!cpus_to_check)
+		cpus_to_check = 1;
+
+	for (i = 0; i < cpus_to_check; ++i) {
+		atomic_t *nmbr_entries_ptr = per_cpu_ptr(&rds_cfu_entries, gc_control.next_cpu);
+		union lfstack *stack = per_cpu_ptr(&rds_cfu_cache, gc_control.next_cpu);
+		unsigned int nmbr_cleaned = 0;
+		struct lfstack_el *el;
+
+		while ((el = lfstack_pop(stack))) {
+			struct rds_cfu_cache_entry *entry =
+				container_of(el, struct rds_cfu_cache_entry, list);
+
+			++nmbr_cleaned;
+			rds_page_free(entry->pg);
+		}
+
+		atomic_sub(nmbr_cleaned, nmbr_entries_ptr);
+		rds_stats_add(s_copy_from_user_cache_get, nmbr_cleaned);
+		if (++gc_control.next_cpu >= num_possible_cpus())
+			gc_control.next_cpu = 0;
+	}
+}
+
+static void rds_cfu_cache_gc_worker(struct work_struct *work)
+{
+	rds_cfu_cache_do_gc();
+
+	/* To pair with smp_store_release() below */
+	if (!smp_load_acquire(&rds_cfu_cache_tearing_down))
+		rds_queue_delayed_work(NULL, rds_wq, &gc_control.work,
+				       msecs_to_jiffies(rds_cfu_cache_gc_interval * 1000),
+				       "CFU_Cache_gc");
+}
+
+void rds_cfu_init_cache(void)
+{
+	INIT_DELAYED_WORK(&gc_control.work, rds_cfu_cache_gc_worker);
+
+	gc_control.percent_cpus_to_clean = 10;
+	gc_control.next_cpu = 0;
+	rds_queue_delayed_work(NULL, rds_wq, &gc_control.work,
+			       msecs_to_jiffies(rds_cfu_cache_gc_interval * 1000),
+			       "CFU_Cache_gc");
+}
+
+void rds_cfu_fini_cache(void)
+{
+	/* To pair with the smp_load_acquire() above */
+	smp_store_release(&rds_cfu_cache_tearing_down, true);
+	cancel_delayed_work_sync(&gc_control.work);
+
+	gc_control.percent_cpus_to_clean = 100;
+	rds_cfu_cache_do_gc();
+}
+
 /*
  * This relies on dma_map_sg() not touching sg[].page during merging.
  */
 static void rds_message_purge(struct rds_message *rm)
 {
+	atomic_t *nmbr_entries_ptr = per_cpu_ptr(&rds_cfu_entries, rm->m_alloc_cpu);
+	union lfstack *stack = per_cpu_ptr(&rds_cfu_cache, rm->m_alloc_cpu);
+	struct lfstack_el *first = NULL;
+	unsigned int cache_puts = 0;
+	struct lfstack_el *last;
 	unsigned long i;
 
 	if (unlikely(test_bit(RDS_MSG_PAGEVEC, &rm->m_flags)))
 		return;
 
 	for (i = 0; i < rm->data.op_nents; i++) {
-		rdsdebug("putting data page %p\n", (void *)sg_page(&rm->data.op_sg[i]));
-		/* XXX will have to put_page for page refs */
-		rds_page_free(sg_page(&rm->data.op_sg[i]));
+		if (rm->m_alloc_cpu != NUMA_NO_NODE && rm->data.op_sg[i].length >= PAGE_SIZE &&
+		    atomic_read(nmbr_entries_ptr) < rds_sysctl_cfu_cache_cap) {
+			struct rds_cfu_cache_entry *entry =
+				page_address(sg_page(rm->data.op_sg + i));
+
+			++cache_puts;
+			if (!first) {
+				first = &entry->list;
+				last = first;
+			} else {
+				lfstack_link(last, &entry->list);
+				last = lfstack_next(last);
+			}
+			entry->pg = sg_page(rm->data.op_sg + i);
+			last->next = NULL;
+		} else {
+			rds_page_free(sg_page(&rm->data.op_sg[i]));
+		}
+	}
+	if (first) {
+		lfstack_push_many(stack, first, last);
+		atomic_add(cache_puts, nmbr_entries_ptr);
+		rds_stats_add(s_copy_from_user_cache_put, cache_puts);
 	}
+
 	rm->data.op_nents = 0;
 
 	if (rm->rdma.op_active)
@@ -272,6 +377,7 @@ struct rds_message *rds_message_alloc(unsigned int extra_len, gfp_t gfp)
 
 	rm->m_used_sgs = 0;
 	rm->m_total_sgs = extra_len / sizeof(struct scatterlist);
+	rm->m_alloc_cpu = NUMA_NO_NODE;
 
 	atomic_set(&rm->m_refcount, 1);
 	INIT_LIST_HEAD(&rm->m_sock_item);
@@ -319,12 +425,34 @@ int rds_message_copy_from_user(struct rds_message *rm, struct iov_iter *from)
 
 	while (iov_iter_count(from)) {
 		if (!sg_page(sg)) {
-			ret = rds_page_remainder_alloc(sg, iov_iter_count(from),
-						       GFP_HIGHUSER, NUMA_NO_NODE);
-			if (ret)
-				return ret;
+			if (iov_iter_count(from) >= PAGE_SIZE) {
+				union lfstack *stack = per_cpu_ptr(&rds_cfu_cache,
+								    smp_processor_id());
+				struct lfstack_el *el = lfstack_pop(stack);
+
+				if (el) {
+					atomic_t *nmbr_entries_ptr =
+						per_cpu_ptr(&rds_cfu_entries, smp_processor_id());
+					struct rds_cfu_cache_entry *entry =
+						container_of(el, struct rds_cfu_cache_entry, list);
+
+					sg_set_page(sg, entry->pg, PAGE_SIZE, 0);
+					rds_stats_inc(s_copy_from_user_cache_get);
+					atomic_dec(nmbr_entries_ptr);
+				}
+			}
+
+			if (!sg_page(sg)) {
+				ret = rds_page_remainder_alloc(sg, iov_iter_count(from),
+							       GFP_HIGHUSER, NUMA_NO_NODE);
+				if (ret)
+					return ret;
+			}
+
 			rm->data.op_nents++;
 			sg_off = 0;
+			if (rm->m_alloc_cpu == NUMA_NO_NODE)
+				rm->m_alloc_cpu = smp_processor_id();
 		}
 
 		to_copy = min_t(unsigned long, iov_iter_count(from),

net/rds/rds.h

@@ -725,6 +725,7 @@ struct rds_message {
 
 	struct rds_conn_path *m_conn_path;
 	struct rds_csum m_payload_csum;
+	int m_alloc_cpu;
 };
 
 /*
@@ -1036,6 +1037,8 @@ struct rds_statistics {
 	uint64_t	s_page_remainder_miss;
 	uint64_t	s_copy_to_user;
 	uint64_t	s_copy_from_user;
+	uint64_t	s_copy_from_user_cache_get;
+	uint64_t	s_copy_from_user_cache_put;
 	uint64_t	s_cong_update_queued;
 	uint64_t	s_cong_update_received;
 	uint64_t	s_cong_send_error;
@@ -1363,6 +1366,8 @@ void rds_message_addref(struct rds_message *rm);
 void rds_message_put(struct rds_message *rm);
 void rds_message_wait(struct rds_message *rm);
 void rds_message_unmapped(struct rds_message *rm);
+void rds_cfu_init_cache(void);
+void rds_cfu_fini_cache(void);
 
 static inline void rds_message_make_checksum(struct rds_header *hdr)
 {
@@ -1492,6 +1497,8 @@ extern unsigned int  rds_sysctl_conn_hb_timeout;
 extern unsigned int  rds_sysctl_conn_hb_interval;
 extern unsigned long rds_sysctl_dr_sock_cancel_jiffies;
 extern unsigned int  rds_sysctl_enable_payload_csum;
+extern unsigned int  rds_sysctl_cfu_cache_cap;
+extern unsigned int  rds_cfu_cache_gc_interval;
 
 /* threads.c */
 int rds_threads_init(void);

net/rds/stats.c

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2006 Oracle.  All rights reserved.
+ * Copyright (c) 2006, 2023, Oracle and/or its affiliates.
  *
  * This software is available to you under a choice of one of two
  * licenses.  You may choose to be licensed under the terms of the GNU
@@ -87,6 +87,8 @@ static char *rds_stat_names[] = {
 	"page_remainder_miss",
 	"copy_to_user",
 	"copy_from_user",
+	"copy_from_user_cache_get",
+	"copy_from_user_cache_put",
 	"cong_update_queued",
 	"cong_update_received",
 	"cong_send_error",

net/rds/sysctl.c

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2006, 2019 Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2006, 2023, Oracle and/or its affiliates.
  *
  * This software is available to you under a choice of one of two
  * licenses.  You may choose to be licensed under the terms of the GNU
@@ -88,6 +88,10 @@ static int rds_sysctl_sol_rds = SOL_RDS;
 
 unsigned int rds_sysctl_enable_payload_csum;
 
+unsigned int rds_sysctl_cfu_cache_cap = 512;
+
+unsigned int rds_cfu_cache_gc_interval = 1;
+
 static struct ctl_table rds_sysctl_rds_table[] = {
 	{
 		.procname       = "reconnect_min_delay_ms",
@@ -204,6 +208,20 @@ static struct ctl_table rds_sysctl_rds_table[] = {
 		.mode           = 0644,
 		.proc_handler   = proc_douintvec,
 	},
+	{
+		.procname       = "copy_from_user_per_cpu_cache_cap",
+		.data           = &rds_sysctl_cfu_cache_cap,
+		.maxlen         = sizeof(rds_sysctl_cfu_cache_cap),
+		.mode           = 0644,
+		.proc_handler   = proc_douintvec,
+	},
+	{
+		.procname       = "copy_from_user_gc_interval_secs",
+		.data           = &rds_cfu_cache_gc_interval,
+		.maxlen         = sizeof(rds_cfu_cache_gc_interval),
+		.mode           = 0644,
+		.proc_handler   = proc_douintvec,
+	},
 };
 
 void rds_sysctl_exit(void)