prov/shm: new shm architecture (updated)

fabtests/pytest/efa/test_rdm.py

@@ -248,7 +248,7 @@ def test_implicit_av(cmdline_args, unexpected_path, msg_size):
     if not os.path.exists(os.path.join(binpath, "fi_efa_implicit_av_test")):
         pytest.skip("implicit AV test not found")
 
-    if (cmdline_args.server_id == cmdline_args.client_id) and unexpected_path and msg_size > 1024:
+    if (cmdline_args.server_id == cmdline_args.client_id) and unexpected_path and msg_size >= 1024:
         pytest.skip("SHM provider will use CMA protocol needs test modifications")
 
     test_cmd = f"fi_efa_implicit_av_test -L -c 5 -S {msg_size}"

include/ofi.h

@@ -217,6 +217,8 @@ ofi_poll_del(struct fid_poll *pollset, struct fid *event_fid, uint64_t flags)
 		_a > _b ? _a : _b; })
 #endif
 
+#define MIN3(a, b, c) MIN(MIN(a, b), c)
+
 #define ofi_div_ceil(a, b) ((a + b - 1) / b)
 
 static inline int ofi_val64_gt(uint64_t x, uint64_t y) {

include/ofi_atomic_queue.h

@@ -94,6 +94,7 @@ extern "C" {
 
 #define OFI_CACHE_LINE_SIZE (64)
 
+typedef void (*ofi_aq_init_fn)(void *);
 /*
  * Base address of atomic queue must be cache line aligned to maximize atomic
  * value perforamnce benefits
@@ -110,36 +111,46 @@ struct name {							\
 	uint8_t		pad0[OFI_CACHE_LINE_SIZE -		\
 			     sizeof(ofi_atomic64_t)];		\
 	ofi_atomic64_t	read_pos;				\
+	ofi_aq_init_fn	init_fn;				\
 	uint8_t		pad1[OFI_CACHE_LINE_SIZE -		\
-			     sizeof(ofi_atomic64_t)];		\
+			     (sizeof(ofi_atomic64_t) +		\
+			     sizeof(ofi_aq_init_fn))];		\
 	int		size;					\
 	int		size_mask;				\
 	uint8_t		pad2[OFI_CACHE_LINE_SIZE -		\
 			     (sizeof(int) * 2)];		\
 	struct name ## _entry entry[];				\
 } __attribute__((__aligned__(64)));				\
 								\
-static inline void name ## _init(struct name *aq, size_t size)	\
+static inline void name ## _init(struct name *aq, size_t size,	\
+				 ofi_aq_init_fn init_fn)	\
 {								\
 	size_t i;						\
 	assert(size == roundup_power_of_two(size));		\
 	assert(!((uintptr_t) aq % OFI_CACHE_LINE_SIZE));	\
 	aq->size = size;					\
 	aq->size_mask = aq->size - 1;				\
+	aq->init_fn = init_fn;					\
 	ofi_atomic_initialize64(&aq->write_pos, 0);		\
 	ofi_atomic_initialize64(&aq->read_pos, 0);		\
-	for (i = 0; i < size; i++)				\
+	for (i = 0; i < size; i++) {				\
 		ofi_atomic_initialize64(&aq->entry[i].seq, i);	\
+		if (aq->init_fn)				\
+			aq->init_fn(&aq->entry[i].buf);		\
+		aq->entry[i].noop = false;			\
+	}							\
 }								\
 								\
 static inline struct name * name ## _create(size_t size)	\
 {								\
 	struct name *aq;					\
-	aq = (struct name*) calloc(1, sizeof(*aq) +		\
-		sizeof(struct name ## _entry) *			\
-		(roundup_power_of_two(size)));			\
+	aq = (struct name *) aligned_alloc(			\
+			OFI_CACHE_LINE_SIZE, sizeof(*aq) +	\
+			sizeof(struct name ## _entry) *		\
+			(roundup_power_of_two(size)));		\
 	if (aq)							\
-		name ##_init(aq, roundup_power_of_two(size));	\
+		name ##_init(aq, roundup_power_of_two(size),	\
+			     NULL);				\
 	return aq;						\
 }								\
 								\
@@ -181,9 +192,11 @@ static inline void name ## _release(struct name *aq,		\
 {								\
 	struct name ## _entry *ce;				\
 	ce = container_of(buf, struct name ## _entry, buf);	\
+	if (aq->init_fn)					\
+		aq->init_fn(&ce->buf);				\
 	ofi_atomic_store_explicit64(&ce->seq,			\
-			      pos + aq->size,			\
-			      memory_order_release);		\
+				    pos + aq->size,		\
+				    memory_order_release);	\
 }								\
 static inline int name ## _head(struct name *aq,		\
 		entrytype **buf, int64_t *pos)			\

include/ofi_hmem.h

@@ -437,6 +437,9 @@ ssize_t ofi_copy_from_mr_iov(void *dest, size_t size, struct ofi_mr **mr,
 ssize_t ofi_copy_to_mr_iov(struct ofi_mr **mr, const struct iovec *iov,
 			   size_t iov_count, uint64_t iov_offset,
 			   const void *src, size_t size);
+ssize_t ofi_copy_mr_iov(struct ofi_mr **mr, const struct iovec *iov,
+			size_t iov_count, size_t offset, void *buf,
+			size_t size, int dir);
 
 int ofi_hmem_get_handle(enum fi_hmem_iface iface, void *base_addr,
 			size_t size, void **handle);

include/ofi_mem.h

@@ -259,6 +259,14 @@ static inline void smr_freestack_push_by_offset(struct smr_freestack *fs,
                         fs->object_size);
 }
 
+/* Get entry index in fs */
+static inline int16_t smr_freestack_get_index(struct smr_freestack *fs,
+					      char *local_p)
+{
+	uint64_t offset = ((char*) local_p - (char*) fs);
+	return (offset - fs->entry_base_offset) / fs->object_size;
+}
+
 /* Push by object */
 static inline void smr_freestack_push(struct smr_freestack *fs, void *local_p)
 {
@@ -318,6 +326,12 @@ static inline void* smr_freestack_pop(struct smr_freestack *fs)
 {
 	return (void *) ( ((char*)fs) + smr_freestack_pop_by_offset(fs) );
 }
+
+static inline int16_t smr_freestack_avail(struct smr_freestack *fs)
+{
+	return fs->free;
+}
+
 /*
  * Buffer Pool
  */

include/ofi_xpmem.h

@@ -48,7 +48,7 @@ typedef int64_t xpmem_segid_t;
 #endif /* HAVE_XPMEM */
 
 struct ofi_xpmem_client {
-	uint8_t cap;
+	bool avail;
 	xpmem_apid_t apid;
 	uintptr_t addr_max;
 };

man/fi_shm.7.md

@@ -152,11 +152,6 @@ structures.
 
 The *shm* provider checks for the following environment variables:
 
-*FI_SHM_SAR_THRESHOLD*
-: Maximum message size to use segmentation protocol before switching
-  to mmap (only valid when CMA is not available).  Default: SIZE_MAX
-  (18446744073709551615)
-
 *FI_SHM_TX_SIZE*
 : Maximum number of outstanding tx operations.  Default 1024
 
@@ -181,6 +176,14 @@ The *shm* provider checks for the following environment variables:
     chunks.  This environment variable is provided to fine tune performance
     on different systems.  Default 262144
 
+*FI_SHM_BUFFER_THRESHOLD*
+ :  When to start requesting forced unexpected messaging buffering. When this
+    threshold is reached, the sender will notify the receiver to force buffering
+    of the entire message if it is unexpected. If the message is matched when
+    received, it has no effect. Requesting unexpected message buffering allows
+    shm to support unlimited unexpected messaging (memory permitting).
+    Default: 1
+
 # SEE ALSO
 
 [`fabric`(7)](fabric.7.html),