Define monitor unsubscribe_on_delete

Makefile.am

@@ -92,6 +92,8 @@ common_srcs =				\
 	prov/util/src/ze_ipc_monitor.c	\
 	prov/util/src/xpmem_monitor.c	\
         prov/util/src/kdreg2_mem_monitor.c \
+	prov/util/src/uffd_mem_monitor.c \
+	prov/util/src/import_mem_monitor.c \
 	prov/util/src/util_profile.c \
 	prov/coll/src/coll_attr.c	\
 	prov/coll/src/coll_av.c		\

include/ofi_mr.h

@@ -118,9 +118,12 @@ static inline uint64_t ofi_mr_get_prov_mode(uint32_t version,
 	}
 }
 
-
 /* Single lock used by all memory monitors and MR caches. */
 extern pthread_mutex_t mm_lock;
+
+/* Lock used to coordinate monitor states. */
+extern pthread_mutex_t mm_state_lock;
+
 /* The read-write lock is an additional lock used to protect the dlist_entry
  * list of ofi_mem_monitor. Due to the necessity of releasing the mm_lock
  * while walking the dlist in ofi_monitor_notify, we need a separate lock to

libfabric.vcxproj

@@ -760,6 +760,8 @@
     <ClCompile Include="prov\util\src\rocr_ipc_monitor.c" />
     <ClCompile Include="prov\util\src\xpmem_monitor.c" />
     <ClCompile Include="prov\util\src\kdreg2_mem_monitor.c" />
+    <ClCompile Include="prov\util\src\uffd_mem_monitor.c" />
+    <ClCompile Include="prov\util\src\import_mem_monitor.c" />
     <ClCompile Include="prov\coll\src\coll_attr.c" />
     <ClCompile Include="prov\coll\src\coll_av.c" />
     <ClCompile Include="prov\coll\src\coll_av_set.c" />

prov/util/src/import_mem_monitor.c

@@ -0,0 +1,195 @@
+/*
+ * Copyright (c) 2017 Cray Inc. All rights reserved.
+ * Copyright (c) 2017-2021 Intel Inc. All rights reserved.
+ * Copyright (c) 2019-2021 Amazon.com, Inc. or its affiliates.
+ *                         All rights reserved.
+ * (C) Copyright 2020,2024 Hewlett Packard Enterprise Development LP
+ * Copyright (C) 2024 Cornelis Networks. All rights reserved.
+ *
+ * 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
+ * General Public License (GPL) Version 2, available from the file
+ * COPYING in the main directory of this source tree, or the
+ * BSD license below:
+ *
+ *     Redistribution and use in source and binary forms, with or
+ *     without modification, are permitted provided that the following
+ *     conditions are met:
+ *
+ *      - Redistributions of source code must retain the above
+ *        copyright notice, this list of conditions and the following
+ *        disclaimer.
+ *
+ *      - Redistributions in binary form must reproduce the above
+ *        copyright notice, this list of conditions and the following
+ *        disclaimer in the documentation and/or other materials
+ *        provided with the distribution.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
+ * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+ * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
+ * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS
+ * BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
+ * ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
+ * CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+#include <unistd.h>
+
+#include <ofi_mr.h>
+#include <ofi_enosys.h>
+#include <rdma/fi_ext.h>
+
+static void ofi_import_monitor_init(struct ofi_mem_monitor *monitor);
+static void ofi_import_monitor_cleanup(struct ofi_mem_monitor *monitor);
+static int ofi_import_monitor_start(struct ofi_mem_monitor *monitor);
+static void ofi_import_monitor_stop(struct ofi_mem_monitor *monitor);
+static int ofi_import_monitor_subscribe(struct ofi_mem_monitor *notifier,
+					const void *addr, size_t len,
+					union ofi_mr_hmem_info *hmem_info);
+static void ofi_import_monitor_unsubscribe(struct ofi_mem_monitor *notifier,
+					   const void *addr, size_t len,
+					   union ofi_mr_hmem_info *hmem_info);
+static bool ofi_import_monitor_valid(struct ofi_mem_monitor *notifier,
+				     const struct ofi_mr_info *info,
+				     struct ofi_mr_entry *entry);
+
+struct ofi_import_monitor {
+	struct ofi_mem_monitor monitor;
+	struct fid_mem_monitor *impfid;
+};
+
+static struct ofi_import_monitor impmon = {
+	.monitor.iface = FI_HMEM_SYSTEM,
+	.monitor.init = ofi_import_monitor_init,
+	.monitor.cleanup = ofi_import_monitor_cleanup,
+	.monitor.start = ofi_import_monitor_start,
+	.monitor.stop = ofi_import_monitor_stop,
+	.monitor.subscribe = ofi_import_monitor_subscribe,
+	.monitor.unsubscribe = ofi_import_monitor_unsubscribe,
+	.monitor.valid = ofi_import_monitor_valid,
+	.monitor.name = "import",
+};
+
+struct ofi_mem_monitor *import_monitor = &impmon.monitor;
+
+static void ofi_import_monitor_init(struct ofi_mem_monitor *monitor)
+{
+	ofi_monitor_init(monitor);
+}
+
+static void ofi_import_monitor_cleanup(struct ofi_mem_monitor *monitor)
+{
+	assert(!impmon.impfid);
+	ofi_monitor_cleanup(monitor);
+}
+
+static int ofi_import_monitor_start(struct ofi_mem_monitor *monitor)
+{
+	if (!impmon.impfid)
+		return -FI_ENOSYS;
+
+	return impmon.impfid->export_ops->start(impmon.impfid);
+}
+
+static void ofi_import_monitor_stop(struct ofi_mem_monitor *monitor)
+{
+	assert(impmon.impfid);
+	impmon.impfid->export_ops->stop(impmon.impfid);
+}
+
+static int ofi_import_monitor_subscribe(struct ofi_mem_monitor *notifier,
+					const void *addr, size_t len,
+					union ofi_mr_hmem_info *hmem_info)
+{
+	assert(impmon.impfid);
+	return impmon.impfid->export_ops->subscribe(impmon.impfid, addr, len);
+}
+
+static void ofi_import_monitor_unsubscribe(struct ofi_mem_monitor *notifier,
+					   const void *addr, size_t len,
+					   union ofi_mr_hmem_info *hmem_info)
+{
+	assert(impmon.impfid);
+	impmon.impfid->export_ops->unsubscribe(impmon.impfid, addr, len);
+}
+
+static bool ofi_import_monitor_valid(struct ofi_mem_monitor *notifier,
+				     const struct ofi_mr_info *info,
+				     struct ofi_mr_entry *entry)
+{
+	assert(impmon.impfid);
+	return impmon.impfid->export_ops->valid(impmon.impfid,
+						entry->info.iov.iov_base,
+						entry->info.iov.iov_len);
+}
+
+static void ofi_import_monitor_notify(struct fid_mem_monitor *monitor,
+				      const void *addr, size_t len)
+{
+	assert(monitor->fid.context == &impmon);
+	pthread_rwlock_rdlock(&mm_list_rwlock);
+	pthread_mutex_lock(&mm_lock);
+	ofi_monitor_notify(&impmon.monitor, addr, len);
+	pthread_mutex_unlock(&mm_lock);
+	pthread_rwlock_unlock(&mm_list_rwlock);
+}
+
+static int ofi_close_import(struct fid *fid)
+{
+	pthread_mutex_lock(&mm_state_lock);
+	impmon.monitor.state = FI_MM_STATE_IDLE;
+	pthread_mutex_unlock(&mm_state_lock);
+	impmon.impfid = NULL;
+	return 0;
+}
+
+static struct fi_ops_mem_notify import_ops = {
+	.size = sizeof(struct fi_ops_mem_notify),
+	.notify = ofi_import_monitor_notify,
+};
+
+static struct fi_ops impfid_ops = {
+	.size = sizeof(struct fi_ops),
+	.close = ofi_close_import,
+	.bind = fi_no_bind,
+	.control = fi_no_control,
+	.ops_open = fi_no_ops_open,
+	.tostr = fi_no_tostr,
+	.ops_set = fi_no_ops_set,
+};
+
+int ofi_monitor_import(struct fid *fid)
+{
+	struct fid_mem_monitor *impfid;
+
+	if (fid->fclass != FI_CLASS_MEM_MONITOR)
+		return -FI_ENOSYS;
+
+	if (impmon.impfid) {
+		FI_WARN(&core_prov, FI_LOG_MR,
+			"imported monitor already exists\n");
+		return -FI_EBUSY;
+	}
+
+	if (default_monitor && !dlist_empty(&default_monitor->list)) {
+		FI_WARN(&core_prov, FI_LOG_MR,
+			"cannot replace active monitor\n");
+		return -FI_EBUSY;
+	}
+
+	impfid = container_of(fid, struct fid_mem_monitor, fid);
+	if (impfid->export_ops->size < sizeof(struct fi_ops_mem_monitor))
+		return -FI_EINVAL;
+
+	impmon.impfid = impfid;
+	impfid->fid.context = &impmon;
+	impfid->fid.ops = &impfid_ops;
+	impfid->import_ops = &import_ops;
+
+	FI_INFO(&core_prov, FI_LOG_MR,
+		"setting imported memory monitor as default\n");
+	default_monitor = &impmon.monitor;
+	return 0;
+}

prov/util/src/rocr_mem_monitor.c

@@ -44,13 +44,17 @@
 struct rocr_mm_entry {
 	struct iovec iov;
 	struct ofi_rbnode *node;
+	struct dlist_entry entry;
 };
 
 struct rocr_mm {
 	struct ofi_mem_monitor mm;
 	struct ofi_rbmap *dev_region_tree;
 };
 
+static void rocr_unsubscribe(struct ofi_mem_monitor *monitor, const void *addr,
+			     size_t len, union ofi_mr_hmem_info *hmem_info,
+			     struct dlist_entry *free_list);
 static int rocr_mm_start(struct ofi_mem_monitor *monitor);
 static void rocr_mm_stop(struct ofi_mem_monitor *monitor);
 static int rocr_mm_subscribe(struct ofi_mem_monitor *monitor, const void *addr,
@@ -133,15 +137,22 @@ static struct rocr_mm_entry *rocr_mm_entry_find(const void *addr)
 static void rocr_mm_dealloc_cb(void *addr, void *user_data)
 {
 	size_t len = (size_t) user_data;
+	DEFINE_LIST(free_list);
+	struct rocr_mm_entry *entry;
 
 	pthread_rwlock_rdlock(&mm_list_rwlock);
 	pthread_mutex_lock(&mm_lock);
-	ofi_monitor_unsubscribe(rocr_monitor, addr, len, NULL);
+	rocr_unsubscribe(rocr_monitor, addr, len, NULL, &free_list);
 	pthread_mutex_unlock(&mm_lock);
 	pthread_rwlock_unlock(&mm_list_rwlock);
+
+	while (!dlist_empty(&free_list)) {
+		dlist_pop_front(&free_list, struct rocr_mm_entry, entry, entry);
+		free(entry);
+	}
 }
 
-static void rocr_mm_entry_free(struct rocr_mm_entry *entry)
+static void rocr_mm_entry_delete(struct rocr_mm_entry *entry)
 {
 	hsa_status_t hsa_ret __attribute__((unused));
 
@@ -161,6 +172,11 @@ static void rocr_mm_entry_free(struct rocr_mm_entry *entry)
 	       hsa_ret == HSA_STATUS_ERROR_INVALID_ARGUMENT);
 
 	ofi_rbmap_delete(rocr_mm.dev_region_tree, entry->node);
+}
+
+static void rocr_mm_entry_free(struct rocr_mm_entry *entry)
+{
+	rocr_mm_entry_delete(entry);
 	free(entry);
 }
 
@@ -260,9 +276,9 @@ static void rocr_mm_stop(struct ofi_mem_monitor *monitor)
 	ofi_rbmap_destroy(rocr_mm.dev_region_tree);
 }
 
-static void rocr_mm_unsubscribe(struct ofi_mem_monitor *monitor,
-				const void *addr, size_t len,
-				union ofi_mr_hmem_info *hmem_info)
+static void rocr_unsubscribe(struct ofi_mem_monitor *monitor, const void *addr,
+			     size_t len, union ofi_mr_hmem_info *hmem_info,
+			     struct dlist_entry *free_list)
 {
 	struct rocr_mm_entry *entry;
 	size_t cur_len = len;
@@ -286,7 +302,14 @@ static void rocr_mm_unsubscribe(struct ofi_mem_monitor *monitor,
 
 		next_addr = (void *) ((uintptr_t) ofi_iov_end(&entry->iov) + 1);
 
-		rocr_mm_entry_free(entry);
+		/* Since unsubscribed is called with mm_lock held, calling free
+		 * may result in deadlocks if memhooks is used. To prevent this,
+		 * entries are placed on a list to be freed later.
+		 *
+		 * Entry still needs to be deleted.
+		 */
+		rocr_mm_entry_delete(entry);
+		dlist_insert_tail(&entry->entry, free_list);
 
 		cur_len -= MIN((uintptr_t) next_addr - (uintptr_t) cur_addr,
 			       cur_len);
@@ -299,6 +322,18 @@ static void rocr_mm_unsubscribe(struct ofi_mem_monitor *monitor,
 			addr, len);
 }
 
+/* Since ROCR may have many MR cache entries for the same VA range and
+ * ofi_monitor_unsubscribe() is called for every MR cache entry being freed,
+ * ROCR unsubscribe needs to be a noop. Else, MR cache entries may no longer
+ * be monitored.
+ */
+static void rocr_mm_unsubscribe(struct ofi_mem_monitor *monitor,
+				const void *addr, size_t len,
+				union ofi_mr_hmem_info *hmem_info)
+{
+	/* no-op */
+}
+
 /* Subscribe is designed to monitor entire ROCR memory regions even if the user
  * subscribe region is smaller. All ROCR memory regions are inserted into an RB
  * tree for tracking. Future subscriptions will always reuse RB tree entries if