prov/shm, prov/efa: Allow more than 256 shm peers

fabtests/Makefile.am

@@ -62,6 +62,7 @@ bin_PROGRAMS = \
 	unit/fi_mr_cache_evict \
 	unit/fi_cntr_test \
 	unit/fi_av_test \
+	unit/fi_av_capacity_test \
 	unit/fi_dom_test \
 	unit/fi_getinfo_test \
 	unit/fi_setopt_test \
@@ -208,6 +209,7 @@ nobase_dist_config_DATA = \
 	pytest/shm/conftest.py \
 	pytest/shm/shm_common.py \
 	pytest/shm/test_av.py \
+	pytest/shm/test_av_capacity.py \
 	pytest/shm/test_cntr.py \
 	pytest/shm/test_cq.py \
 	pytest/shm/test_dom.py \
@@ -483,6 +485,11 @@ unit_fi_av_test_SOURCES = \
 	$(unit_srcs)
 unit_fi_av_test_LDADD = libfabtests.la
 
+unit_fi_av_capacity_test_SOURCES = \
+	unit/av_capacity_test.c \
+	$(unit_srcs)
+unit_fi_av_capacity_test_LDADD = libfabtests.la
+
 unit_fi_dom_test_SOURCES = \
 	unit/dom_test.c \
 	$(unit_srcs)
@@ -691,6 +698,7 @@ dummy_man_pages = \
 	man/man1/fi_rdm_tagged_pingpong.1 \
 	man/man1/fi_rma_bw.1 \
 	man/man1/fi_av_test.1 \
+	man/man1/fi_av_capacity_test.1 \
 	man/man1/fi_cntr_test.1 \
 	man/man1/fi_cq_test.1 \
 	man/man1/fi_dom_test.1 \

fabtests/man/fabtests.7.md

@@ -211,6 +211,11 @@ testing scope is limited.
 *fi_av_test*
 : Verify address vector interfaces.
 
+*fi_av_capacity_test*
+: Verify that an address vector accepts at least a given number of addresses,
+  rejects inserts beyond its limit, and accepts an insert again after a
+  remove. Supports providers using the FI_ADDR_STR address format.
+
 *fi_cntr_test*
 : Tests counter creation and destruction.
 

fabtests/man/man1/fi_av_capacity_test.1

@@ -0,0 +1 @@
+.so man7/fabtests.7

fabtests/pytest/shm/test_av_capacity.py

@@ -0,0 +1,12 @@
+import os
+
+import pytest
+
+
+@pytest.mark.pr_ci
+@pytest.mark.unit
+def test_av_capacity(cmdline_args):
+    from common import UnitTest
+    min_addrs = max(256, os.cpu_count() or 0)
+    test = UnitTest(cmdline_args, "fi_av_capacity_test -n " + str(min_addrs))
+    test.run()

fabtests/scripts/runfabtests.sh

@@ -240,6 +240,7 @@ unit_tests=(
 	"fi_getinfo_test -s SERVER_ADDR GOOD_ADDR"
 	"fi_av_test -g GOOD_ADDR -n 1 -s SERVER_ADDR -e rdm"
 	"fi_av_test -g GOOD_ADDR -n 1 -s SERVER_ADDR -e dgram"
+	"fi_av_capacity_test -n 256"
 	"fi_dom_test -n 2"
 	"fi_eq_test"
 	"fi_cq_test"

fabtests/unit/av_capacity_test.c

@@ -0,0 +1,206 @@
+/* SPDX-License-Identifier: BSD-2-Clause OR GPL-2.0-only */
+/* SPDX-FileCopyrightText: Copyright Amazon.com, Inc. or its affiliates. All rights reserved. */
+
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <getopt.h>
+#include <inttypes.h>
+
+#include <rdma/fi_cm.h>
+#include <rdma/fi_errno.h>
+
+#include "shared.h"
+#include "unit_common.h"
+
+#define AV_CAPACITY_ADDR_LEN 256
+
+/* 
+ * Distinct from FI_ADDR_NOTAVAIL so that a failed insert which leaves fi_addr
+ * untouched can be told apart from one where the provider correctly set it to
+ * FI_ADDR_NOTAVAIL.
+ */
+#define AV_CAPACITY_ADDR_POISON ((fi_addr_t) 0xdeadbeefdeadbeefULL)
+
+static char base_addr[AV_CAPACITY_ADDR_LEN];
+
+static int insert_one(struct fid_av *test_av, int index, fi_addr_t *fi_addr)
+{
+	char addr[AV_CAPACITY_ADDR_LEN];
+	int len;
+
+	len = snprintf(addr, sizeof(addr), "%s_%d", base_addr, index);
+	if (len < 0 || len >= (int) sizeof(addr))
+		return -FI_ETOOSMALL;
+
+	*fi_addr = AV_CAPACITY_ADDR_POISON;
+	return fi_av_insert(test_av, addr, 1, fi_addr, 0, NULL);
+}
+
+static int get_base_addr(void)
+{
+	size_t len = sizeof(base_addr) - 16;
+	int ret;
+
+	memset(base_addr, 0, sizeof(base_addr));
+	ret = fi_getname(&ep->fid, base_addr, &len);
+	if (ret) {
+		FT_PRINTERR("fi_getname", ret);
+		return ret;
+	}
+	base_addr[sizeof(base_addr) - 1] = '\0';
+	return 0;
+}
+
+static int av_capacity(int min_addrs, int max_addrs)
+{
+	struct fi_av_attr attr = { .type = FI_AV_TABLE };
+	struct fid_av *test_av;
+	fi_addr_t *fi_addrs;
+	fi_addr_t extra;
+	int inserted;
+	int ret, close_ret;
+
+	fi_addrs = calloc(max_addrs, sizeof(*fi_addrs));
+	if (!fi_addrs)
+		return -FI_ENOMEM;
+
+	ret = fi_av_open(domain, &attr, &test_av, NULL);
+	if (ret) {
+		FT_PRINTERR("fi_av_open", ret);
+		goto out;
+	}
+
+	for (inserted = 0; inserted < max_addrs; inserted++) {
+		ret = insert_one(test_av, inserted, &fi_addrs[inserted]);
+		if (ret != 1)
+			break;
+	}
+
+	if (inserted == 0) {
+		printf("cannot insert generated addresses: %d\n", ret);
+		ret = -FI_ENOSYS;
+		goto close;
+	}
+
+	printf("inserted %d addresses\n", inserted);
+	if (inserted < min_addrs) {
+		printf("AV accepted %d addresses, expected at least %d\n",
+		       inserted, min_addrs);
+		ret = -FI_EOTHER;
+		goto close;
+	}
+
+	if (inserted < max_addrs && fi_addrs[inserted] != FI_ADDR_NOTAVAIL) {
+		printf("insert beyond the limit returned %d, fi_addr %" PRIu64
+		       "\n", ret, fi_addrs[inserted]);
+		ret = -FI_EOTHER;
+		goto close;
+	}
+
+	ret = fi_av_remove(test_av, &fi_addrs[0], 1, 0);
+	if (ret) {
+		FT_PRINTERR("fi_av_remove", ret);
+		goto close;
+	}
+
+	ret = insert_one(test_av, inserted, &extra);
+	if (ret != 1 || extra == FI_ADDR_NOTAVAIL ||
+	    extra == AV_CAPACITY_ADDR_POISON) {
+		printf("insert after remove returned %d, fi_addr %" PRIu64 "\n",
+		       ret, extra);
+		ret = -FI_EOTHER;
+		goto close;
+	}
+	ret = 0;
+
+close:
+	close_ret = fi_close(&test_av->fid);
+	if (close_ret) {
+		FT_PRINTERR("fi_close", close_ret);
+		if (!ret)
+			ret = close_ret;
+	}
+out:
+	free(fi_addrs);
+	return ret;
+}
+
+static void usage(char *name)
+{
+	ft_unit_usage(name, "Unit test for AV capacity");
+	FT_PRINT_OPTS_USAGE("-n <num>",
+			    "minimum number of addresses the AV must accept (default 1)");
+	FT_PRINT_OPTS_USAGE("-m <num>",
+			    "maximum number of addresses to insert (default 65536)");
+}
+
+int main(int argc, char **argv)
+{
+	int min_addrs = 1, max_addrs = 65536;
+	int op, ret, cleanup_ret;
+
+	hints = fi_allocinfo();
+	if (!hints)
+		return EXIT_FAILURE;
+
+	while ((op = getopt(argc, argv, FAB_OPTS "n:m:h")) != -1) {
+		switch (op) {
+		case 'n':
+			min_addrs = atoi(optarg);
+			break;
+		case 'm':
+			max_addrs = atoi(optarg);
+			break;
+		default:
+			ft_parseinfo(op, optarg, hints, &opts);
+			break;
+		case '?':
+		case 'h':
+			usage(argv[0]);
+			return EXIT_FAILURE;
+		}
+	}
+
+	if (min_addrs <= 0 || max_addrs <= 0 || min_addrs > max_addrs) {
+		usage(argv[0]);
+		return EXIT_FAILURE;
+	}
+
+	hints->ep_attr->type = FI_EP_RDM;
+	hints->mode = ~0;
+	hints->domain_attr->mode = ~0;
+	hints->domain_attr->mr_mode = ~OFI_MR_DEPRECATED;
+
+	ret = ft_getinfo(hints, &fi);
+	if (ret)
+		goto out;
+
+	if (fi->addr_format != FI_ADDR_STR) {
+		printf("address format %s is not supported\n",
+		       fi_tostr(&fi->addr_format, FI_TYPE_ADDR_FORMAT));
+		ret = -FI_ENOSYS;
+		goto out;
+	}
+
+	ret = ft_open_fabric_res();
+	if (ret)
+		goto out;
+
+	ret = ft_alloc_active_res(fi);
+	if (ret)
+		goto out;
+
+	ret = ft_enable_ep(ep, eq, av, txcq, rxcq, txcntr, rxcntr, rma_cntr);
+	if (ret)
+		goto out;
+
+	ret = get_base_addr();
+	if (ret)
+		goto out;
+
+	ret = av_capacity(min_addrs, max_addrs);
+out:
+	cleanup_ret = ft_free_res();
+	return ft_exit_code(ret ? ret : cleanup_ret);
+}

man/fi_efa.7.md

@@ -1257,6 +1257,7 @@ These OFI runtime parameters apply only to the RDM endpoint.
 
 *FI_EFA_SHM_AV_SIZE*
 : Defines the maximum number of entries in SHM provider's address vector.
+  (Default: number of online CPUs, minimum 256)
 
 *FI_EFA_SHM_MAX_MEDIUM_SIZE*
 : Defines the switch point between small/medium message and large message. The message

prov/efa/src/efa_av.h

@@ -22,7 +22,6 @@
 #endif
 
 #define EFA_MIN_AV_SIZE (16384)
-#define EFA_SHM_MAX_AV_COUNT       (256)
 
 struct efa_rdm_av;
 struct efa_rdm_pke;

prov/efa/src/efa_env.c

@@ -14,7 +14,6 @@ struct efa_env efa_env = {
 	.tx_queue_size = 0,
 	.enable_shm_transfer = 1,
 	.use_zcpy_rx = 1,
-	.shm_av_size = 256,
 	.recvwin_size = EFA_RDM_PEER_DEFAULT_REORDER_BUFFER_SIZE,
 	.ooo_pool_chunk_size = 64,
 	.unexp_pool_chunk_size = 1024,
@@ -75,6 +74,15 @@ static void efa_env_unregistered_param_get(void)
 		efa_env.enable_comp_action = atoi(tmp);
 }
 
+static int efa_env_default_shm_av_size(void)
+{
+	long cores = ofi_sysconf(_SC_NPROCESSORS_ONLN);
+
+	if (cores <= 256)
+		return 256;
+	return (int) cores;
+}
+
 /* @brief Read and store the FI_EFA_* environment variables.
  */
 void efa_env_param_get(void)
@@ -128,6 +136,7 @@ void efa_env_param_get(void)
 	fi_param_get_int(&efa_prov, "tx_queue_size", &efa_env.tx_queue_size);
 	fi_param_get_int(&efa_prov, "enable_shm_transfer", &efa_env.enable_shm_transfer);
 	fi_param_get_int(&efa_prov, "use_zcpy_rx", &efa_env.use_zcpy_rx);
+	efa_env.shm_av_size = efa_env_default_shm_av_size();
 	fi_param_get_int(&efa_prov, "shm_av_size", &efa_env.shm_av_size);
 	fi_param_get_int(&efa_prov, "recvwin_size", &efa_env.recvwin_size);
 	fi_param_get_int(&efa_prov, "readcopy_pool_size", &efa_env.readcopy_pool_size);
@@ -230,7 +239,7 @@ void efa_env_define()
 	fi_param_define(&efa_prov, "use_zcpy_rx", FI_PARAM_INT,
 			"Enables the use of application's receive buffers in place of bounce-buffers when feasible. (Default: 1)");
 	fi_param_define(&efa_prov, "shm_av_size", FI_PARAM_INT,
-			"Defines the maximum number of entries in SHM provider's address vector (Default 128).");
+			"Defines the maximum number of entries in SHM provider's address vector. (Default: number of online CPUs, minimum 256)");
 	fi_param_define(&efa_prov, "recvwin_size", FI_PARAM_INT,
 			"Defines the size of sliding receive window. (Default: 16384)");
 	fi_param_define(&efa_prov, "readcopy_pool_size", FI_PARAM_INT,

prov/efa/src/rdm/efa_rdm_av.c

@@ -1880,20 +1880,16 @@ int efa_rdm_av_open(struct fid_domain *domain_fid, struct fi_av_attr *attr,
 		 * the need of the instances with more CPUs.
 		 */
 		av_attr = *attr;
-		if (efa_env.shm_av_size > EFA_SHM_MAX_AV_COUNT) {
-			ret = -FI_ENOSYS;
-			EFA_WARN(FI_LOG_AV,
-				 "The requested av size is beyond"
-				 " shm supported maximum av size: %s\n",
-				 fi_strerror(-ret));
-			goto err_close_util_av_implicit;
-		}
 		av_attr.count = efa_env.shm_av_size;
 		assert(av_attr.type == FI_AV_TABLE);
 		ret = fi_av_open(rdm_domain->shm_domain, &av_attr,
 				 &rdm_av->shm_rdm_av, context);
-		if (ret)
+		if (ret) {
+			EFA_WARN(FI_LOG_AV,
+				 "Failed to open shm AV with FI_EFA_SHM_AV_SIZE=%d: %s\n",
+				 efa_env.shm_av_size, fi_strerror(-ret));
 			goto err_close_util_av_implicit;
+		}
 	}
 
 	/* the shm AV above stays a table: its fi_addrs never reach the application */

prov/efa/test/efa_unit_tests.c

@@ -20,13 +20,13 @@ static int efa_unit_test_mocks_group_setup(void **state)
 	resource = calloc(1, sizeof(struct efa_resource));
 	*state = resource;
 
-	orig_efa_env = efa_env;
-
 	/* run fi_getinfo to populate g_efa_hmem_info and copy it */
 	fi_getinfo(FI_VERSION(2, 0), NULL, NULL, 0, NULL, &info);
 	memcpy(g_efa_hmem_info_backup, g_efa_hmem_info, sizeof(g_efa_hmem_info));
 	fi_freeinfo(info);
 
+	orig_efa_env = efa_env;
+
 	/* Initialize the random number generator */
 	srand(time(NULL));