prov/shm: modify queue for claim

Signed-off-by: Alexia Ingerson <[email protected]>

Author: aingerson

include/ofi_atomic_queue.h

@@ -95,29 +95,40 @@ extern "C" {
 #define OFI_CACHE_LINE_SIZE (64)
 
 typedef void (*ofi_aq_init_fn)(void *);
+enum {
+	OFI_AQ_FREE = 0,
+	OFI_AQ_READY,
+	OFI_AQ_NOOP,
+};
+
 /*
  * Base address of atomic queue must be cache line aligned to maximize atomic
  * value perforamnce benefits
  */
 #define OFI_DECLARE_ATOMIC_Q(entrytype, name)			\
 struct name ## _entry {						\
-	ofi_atomic64_t	seq;					\
-	bool		noop;					\
+	ofi_atomic64_t	state;					\
 	entrytype	buf;					\
 } __attribute__((__aligned__(64)));				\
 								\
 struct name {							\
 	ofi_atomic64_t	write_pos;				\
 	uint8_t		pad0[OFI_CACHE_LINE_SIZE -		\
 			     sizeof(ofi_atomic64_t)];		\
-	ofi_atomic64_t	read_pos;				\
+	int64_t		read_pos;				\
 	ofi_aq_init_fn	init_fn;				\
 	uint8_t		pad1[OFI_CACHE_LINE_SIZE -		\
-			     (sizeof(ofi_atomic64_t) +		\
+			     (sizeof(int64_t) +			\
 			     sizeof(ofi_aq_init_fn))];		\
+	ofi_atomic64_t	claim_avail;				\
+	uint8_t		pad2[OFI_CACHE_LINE_SIZE -		\
+			     sizeof(ofi_atomic64_t)];		\
+	ofi_atomic64_t	discard_avail;				\
+	uint8_t		pad3[OFI_CACHE_LINE_SIZE -		\
+			     sizeof(ofi_atomic64_t)];		\
 	int		size;					\
 	int		size_mask;				\
-	uint8_t		pad2[OFI_CACHE_LINE_SIZE -		\
+	uint8_t		pad4[OFI_CACHE_LINE_SIZE -		\
 			     (sizeof(int) * 2)];		\
 	struct name ## _entry entry[];				\
 } __attribute__((__aligned__(64)));				\
@@ -132,12 +143,13 @@ static inline void name ## _init(struct name *aq, size_t 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);		\
+	aq->read_pos = 0;					\
+	ofi_atomic_initialize64(&aq->discard_avail, 0);		\
+	ofi_atomic_initialize64(&aq->claim_avail, size);	\
 	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;			\
+		ofi_atomic_initialize64(&aq->entry[i].state, OFI_AQ_FREE);\
 	}							\
 }								\
 								\
@@ -158,96 +170,106 @@ static inline void name ## _free(struct name *aq)		\
 {								\
 	free(aq);						\
 }								\
-static inline int name ## _next(struct name *aq,		\
-		entrytype **buf, int64_t *pos)			\
+static inline bool name ## _claim(struct name *aq)		\
 {								\
-	struct name ## _entry *ce;				\
-	int64_t diff, seq;					\
-	*pos = ofi_atomic_load_explicit64(&aq->write_pos,	\
-				    memory_order_relaxed);	\
-	for (;;) {						\
-		ce = &aq->entry[*pos & aq->size_mask];		\
-		seq = ofi_atomic_load_explicit64(&(ce->seq),	\
-			memory_order_acquire);			\
-		diff = seq - *pos;				\
-		if (diff == 0) {				\
-			if (ofi_atomic_compare_exchange_weak64(	\
-				&aq->write_pos, pos,		\
-				*pos + 1))			\
-				break;				\
-		} else if (diff < 0) {				\
-			return -FI_ENOENT;			\
-		} else {					\
-			*pos = ofi_atomic_load_explicit64(	\
-				&aq->write_pos,			\
-				memory_order_relaxed);		\
-		}						\
+	int64_t avail, discard_avail;				\
+	avail = ofi_atomic_sub_explicit64(&aq->claim_avail, 1,	\
+					  memory_order_relaxed);\
+	if (avail > 0)						\
+		return true;					\
+								\
+	discard_avail = ofi_atomic_load_explicit64(		\
+					&aq->discard_avail,	\
+					memory_order_acquire);	\
+	if (discard_avail) {					\
+		if (!ofi_atomic_compare_exchange_weak64(	\
+					&aq->discard_avail,	\
+					&discard_avail, 0))	\
+			goto out;				\
+		ofi_atomic_add_explicit64(&aq->claim_avail,	\
+					  discard_avail,	\
+					  memory_order_relaxed);\
+	}							\
+out:								\
+	ofi_atomic_add_explicit64(&aq->claim_avail, 1,		\
+				  memory_order_relaxed);	\
+	return false;						\
+								\
+}								\
+static inline entrytype *name ## _assign(struct name *aq)	\
+{								\
+	int64_t pos;						\
+	while (1) {						\
+		pos = ofi_atomic_load_explicit64(		\
+					&aq->write_pos,		\
+					memory_order_acquire);	\
+		if (ofi_atomic_compare_exchange_weak64(		\
+					&aq->write_pos, &pos,	\
+					pos + 1))		\
+			break;					\
 	}							\
-	*buf = &ce->buf;					\
-	return FI_SUCCESS;					\
+	return &aq->entry[pos & aq->size_mask].buf;		\
+}								\
+static inline entrytype *name ## _claim_assign(struct name *aq)	\
+{								\
+	if (name ## _claim(aq)) {				\
+		return name ## _assign(aq);			\
+	}							\
+	return NULL;						\
 }								\
 static inline void name ## _release(struct name *aq,		\
-			entrytype *buf,				\
-			int64_t pos)				\
+				    entrytype *buf)		\
 {								\
+	int64_t state = OFI_AQ_FREE;				\
 	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,		\
+	ofi_atomic_store_explicit64(&ce->state,	state,		\
 				    memory_order_release);	\
+	aq->read_pos++;						\
+}								\
+static inline void name ## _discard(struct name *aq)		\
+{								\
+	ofi_atomic_add_explicit64(&aq->discard_avail, 1,	\
+		memory_order_relaxed);				\
 }								\
-static inline int name ## _head(struct name *aq,		\
-		entrytype **buf, int64_t *pos)			\
+static inline void name ## _release_discard(struct name *aq,	\
+					    entrytype *buf)	\
+{								\
+	name ## _release(aq, buf);				\
+	name ## _discard(aq);					\
+}								\
+static inline entrytype *name ## _head(struct name *aq)		\
 {								\
-	int64_t diff, seq;					\
 	struct name ## _entry *ce;				\
+	int64_t state;						\
 again:								\
-	*pos = ofi_atomic_load_explicit64(&aq->read_pos,	\
-			memory_order_relaxed);			\
-	for (;;) {						\
-		ce = &aq->entry[*pos & aq->size_mask];		\
-		seq = ofi_atomic_load_explicit64(&(ce->seq),	\
-			memory_order_acquire);			\
-		diff = seq - (*pos + 1);			\
-		if (diff == 0) {				\
-			if (ofi_atomic_compare_exchange_weak64(	\
-				&aq->read_pos, pos,		\
-				*pos + 1))			\
-				break;				\
-		} else if (diff < 0) {				\
-			return -FI_ENOENT;			\
-		} else {					\
-			*pos = ofi_atomic_load_explicit64(	\
-				&aq->read_pos,			\
-				memory_order_relaxed);		\
-		}						\
-	}							\
-	*buf = &ce->buf;					\
-	if (ce->noop) {						\
-		ce->noop = false;				\
-		name ##_release(aq, *buf, *pos);		\
+	ce = &aq->entry[aq->read_pos & aq->size_mask];		\
+	state = ofi_atomic_load_explicit64(&ce->state,		\
+					   memory_order_acquire);\
+	if (state == OFI_AQ_FREE)				\
+		return NULL;					\
+	if (state == OFI_AQ_NOOP) {				\
+		name ## _release_discard(aq, &ce->buf);		\
 		goto again;					\
 	}							\
-	return FI_SUCCESS;					\
+	return &ce->buf;					\
 }								\
-static inline void name ## _commit(entrytype *buf,		\
-				int64_t pos)			\
+static inline void name ## _commit(entrytype *buf)		\
 {								\
 	struct name ## _entry *ce;				\
+	int64_t state = OFI_AQ_READY;				\
 	ce = container_of(buf, struct name ## _entry, buf);	\
-	ofi_atomic_store_explicit64(&ce->seq, pos + 1,		\
-			      memory_order_release);		\
+	ofi_atomic_store_explicit64(&ce->state, state,		\
+				    memory_order_release);	\
 }								\
-static inline void name ## _discard(entrytype *buf,		\
-				int64_t pos)			\
+static inline void name ## _cancel(entrytype *buf)		\
 {								\
 	struct name ## _entry *ce;				\
 	ce = container_of(buf, struct name ## _entry, buf);	\
-	ce->noop = true;					\
-	ofi_atomic_store_explicit64(&ce->seq, pos + 1,		\
-			      memory_order_release);		\
+	ofi_atomic_store_explicit64(&ce->state, OFI_AQ_NOOP,	\
+				    memory_order_release);	\
 }								\
 void dummy ## name (void) /* work-around global ; scope */
 

prov/shm/src/smr.h

@@ -50,7 +50,6 @@ struct smr_ep {
 	struct ofi_bufpool	*unexp_buf_pool;
 	struct ofi_bufpool	*pend_pool;
 
-	struct slist		overflow_list;
 	struct dlist_entry	sar_list;
 	struct dlist_entry	async_cpy_list;
 	struct dlist_entry	unexp_cmd_list;
@@ -121,13 +120,11 @@ static inline void smr_return_cmd(struct smr_ep *ep, struct smr_cmd *cmd)
 {
 	struct smr_region *peer_smr = smr_peer_region(ep, cmd->hdr.rx_id);
 	uintptr_t peer_ptr;
-	int64_t pos;
 	struct smr_return_entry *queue_entry;
-	int ret;
 
-	ret = smr_return_queue_next(smr_return_queue(peer_smr), &queue_entry,
-				    &pos);
-	if (ret == -FI_ENOENT) {
+	//return queue has built in claim
+	queue_entry = smr_return_queue_assign(smr_return_queue(peer_smr));
+	if (!queue_entry) {
 		/* return queue runs in parallel to command stack
 		 * ie we will never run out of space
 		 */
@@ -141,9 +138,11 @@ static inline void smr_return_cmd(struct smr_ep *ep, struct smr_cmd *cmd)
 	       peer_smr->total_size);
 	queue_entry->ptr = peer_ptr;
 
-	smr_return_queue_commit(queue_entry, pos);
+	smr_return_queue_commit(queue_entry);
 }
 
+void smr_resend_cmd(struct smr_ep *ep, struct smr_cmd *cmd);
+
 struct smr_env {
 	int	disable_cma;
 	int	use_dsa_sar;
@@ -339,8 +338,6 @@ static inline struct smr_freestack *smr_pend_sar_pool(
 void smr_free_sar_bufs(struct smr_ep *ep, struct smr_cmd *cmd,
 		       struct smr_pend_entry *pending);
 
-void smr_try_send_cmd(struct smr_ep *ep, struct smr_cmd *cmd);
-
 int smr_unexp_start(struct fi_peer_rx_entry *rx_entry);
 
 void smr_progress_async(struct smr_ep *ep);

prov/shm/src/smr_atomic.c

@@ -175,7 +175,7 @@ static ssize_t smr_generic_atomic(
 	struct iovec compare_iov[SMR_IOV_LIMIT];
 	struct iovec result_iov[SMR_IOV_LIMIT];
 	uint16_t smr_flags = 0;
-	int64_t tx_id, rx_id, pos;
+	int64_t tx_id, rx_id;
 	int proto;
 	ssize_t ret = -FI_EAGAIN;
 	size_t total_len;
@@ -196,8 +196,8 @@ static ssize_t smr_generic_atomic(
 	if (smr_peer_data(ep->region)[tx_id].sar_status)
 		goto unlock;
 
-	ret = smr_cmd_queue_next(smr_cmd_queue(peer_smr), &ce, &pos);
-	if (ret == -FI_ENOENT) {
+	ce = smr_cmd_queue_claim_assign(smr_cmd_queue(peer_smr));
+	if (!ce) {
 		ret = -FI_EAGAIN;
 		goto unlock;
 	}
@@ -240,7 +240,7 @@ static ssize_t smr_generic_atomic(
 				     total_len, cmd);
 	} else {
 		if (smr_freestack_isempty(smr_cmd_stack(ep->region))) {
-			smr_cmd_queue_discard(ce, pos);
+			smr_cmd_queue_cancel(ce);
 			ret = -FI_EAGAIN;
 			goto unlock;
 		}
@@ -258,7 +258,7 @@ static ssize_t smr_generic_atomic(
 					   compare_iov, compare_count,
 					   total_len, context, smr_flags, cmd);
 		if (ret) {
-			smr_cmd_queue_discard(ce, pos);
+			smr_cmd_queue_cancel(ce);
 			goto unlock;
 		}
 	}
@@ -272,7 +272,7 @@ static ssize_t smr_generic_atomic(
 	}
 
 	smr_format_rma_ioc(cmd, rma_ioc, rma_count);
-	smr_cmd_queue_commit(ce, pos);
+	smr_cmd_queue_commit(ce);
 unlock:
 	ofi_genlock_unlock(&ep->util_ep.lock);
 	return ret;
@@ -350,7 +350,7 @@ static ssize_t smr_atomic_inject(
 	struct smr_region *peer_smr;
 	struct iovec iov;
 	struct fi_rma_ioc rma_ioc;
-	int64_t id, peer_id, pos;
+	int64_t id, peer_id;
 	ssize_t ret = -FI_EAGAIN;
 	size_t total_len;
 	int proto;
@@ -368,8 +368,9 @@ static ssize_t smr_atomic_inject(
 	if (smr_peer_data(ep->region)[id].sar_status)
 		goto unlock;
 
-	ret = smr_cmd_queue_next(smr_cmd_queue(peer_smr), &ce, &pos);
-	if (ret == -FI_ENOENT) {
+
+	ce = smr_cmd_queue_claim_assign(smr_cmd_queue(peer_smr));
+	if (!ce) {
 		ret = -FI_EAGAIN;
 		goto unlock;
 	}
@@ -393,7 +394,7 @@ static ssize_t smr_atomic_inject(
 	} else {
 		proto = smr_proto_inject;
 		if (smr_freestack_isempty(smr_cmd_stack(ep->region))) {
-			smr_cmd_queue_discard(ce, pos);
+			smr_cmd_queue_cancel(ce);
 			ret = -FI_EAGAIN;
 			goto unlock;
 		}
@@ -407,13 +408,13 @@ static ssize_t smr_atomic_inject(
 					   &iov, 1, NULL, NULL, 0, NULL, NULL,
 					   0, total_len, NULL, 0, cmd);
 		if (ret) {
-			smr_cmd_queue_discard(ce, pos);
+			smr_cmd_queue_cancel(ce);
 			goto unlock;
 		}
 	}
 
 	smr_format_rma_ioc(cmd, &rma_ioc, 1);
-	smr_cmd_queue_commit(ce, pos);
+	smr_cmd_queue_commit(ce);
 
 	if (proto == smr_proto_inline)
 		ofi_ep_peer_tx_cntr_inc(&ep->util_ep, ofi_op_atomic);

prov/shm/src/smr_dsa.c

@@ -645,7 +645,7 @@ static void dsa_complete_tx_work(struct smr_ep *ep, struct smr_pend_entry *pend)
 			ofi_buf_free(pend);
 			return;
 		} else {
-			smr_try_send_cmd(ep, pend->cmd);
+			smr_resend_cmd(ep, pend->cmd);
 		}
 	}