/*
 * Copyright (c) 2013-2018 Intel Corporation. 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 <stdlib.h>
#include <string.h>
#include <sys/uio.h>

#include "ofi_iov.h"
#include "smr.h"


static inline void smr_try_progress_to_sar(struct smr_sar_msg *sar_msg,
				struct smr_resp *resp,
				struct smr_cmd *cmd, struct iovec *iov,
				size_t iov_count, size_t *bytes_done, int *next)
{
	while (*bytes_done < cmd->msg.hdr.size &&
	       smr_copy_to_sar(sar_msg, resp, cmd, iov, iov_count, bytes_done,
			       next));
}

static inline void smr_try_progress_from_sar(struct smr_sar_msg *sar_msg,
				struct smr_resp *resp,
				struct smr_cmd *cmd, struct iovec *iov,
				size_t iov_count, size_t *bytes_done, int *next)
{
	while (*bytes_done < cmd->msg.hdr.size &&
	       smr_copy_from_sar(sar_msg, resp, cmd, iov, iov_count, bytes_done,
				 next));
}

static int smr_progress_resp_entry(struct smr_ep *ep, struct smr_resp *resp,
				   struct smr_tx_entry *pending, uint64_t *err)
{
	struct smr_region *peer_smr;
	size_t inj_offset;
	struct smr_inject_buf *tx_buf = NULL;
	struct smr_sar_msg *sar_msg = NULL;
	uint8_t *src;

	peer_smr = smr_peer_region(ep->region, pending->addr);

	switch (pending->cmd.msg.hdr.op_src) {
	case smr_src_iov:
		break;
	case smr_src_sar:
		sar_msg = smr_get_ptr(peer_smr, pending->cmd.msg.data.sar);
		if (pending->bytes_done == pending->cmd.msg.hdr.size &&
		    sar_msg->sar[0].status == SMR_SAR_FREE &&
		    sar_msg->sar[1].status == SMR_SAR_FREE)
			break;

		if (pending->cmd.msg.hdr.op == ofi_op_read_req)
			smr_try_progress_from_sar(sar_msg, resp,
					&pending->cmd, pending->iov,
				        pending->iov_count, &pending->bytes_done,
					&pending->next);
		else
			smr_try_progress_to_sar(sar_msg, resp,
					&pending->cmd, pending->iov,
					pending->iov_count, &pending->bytes_done,
					&pending->next);
		if (pending->bytes_done != pending->cmd.msg.hdr.size ||
		    sar_msg->sar[0].status != SMR_SAR_FREE ||
		    sar_msg->sar[1].status != SMR_SAR_FREE)
			return -FI_EAGAIN;
		break;
	case smr_src_mmap:
		if (!pending->map_name)
			break;
		if (pending->cmd.msg.hdr.op == ofi_op_read_req) {
			if (!*err) {
				pending->bytes_done = ofi_copy_to_iov(pending->iov,
						pending->iov_count, 0,
						pending->map_ptr,
						pending->cmd.msg.hdr.size);
				if (pending->bytes_done != pending->cmd.msg.hdr.size) {
					FI_WARN(&smr_prov, FI_LOG_EP_CTRL,
						"Incomplete copy from mmapped file\n");
					*err = -FI_EIO;
				}
			}
			munmap(pending->map_ptr, pending->cmd.msg.hdr.size);
		}
		shm_unlink(pending->map_name->name);
		dlist_remove(&pending->map_name->entry);
		free(pending->map_name);
		pending->map_name = NULL;
		break;
	case smr_src_inject:
		inj_offset = (size_t) pending->cmd.msg.hdr.src_data;
		tx_buf = smr_get_ptr(peer_smr, inj_offset);
		if (*err || pending->bytes_done == pending->cmd.msg.hdr.size)
			break;

		src = pending->cmd.msg.hdr.op == ofi_op_atomic_compare ?
		      tx_buf->buf : tx_buf->data;
		pending->bytes_done = ofi_copy_to_iov(pending->iov, pending->iov_count,
				       0, src, pending->cmd.msg.hdr.size);

		if (pending->bytes_done != pending->cmd.msg.hdr.size) {
			FI_WARN(&smr_prov, FI_LOG_EP_CTRL,
				"Incomplete rma read/fetch buffer copied\n");
			*err = FI_EIO;
		}
		break;
	default:
		FI_WARN(&smr_prov, FI_LOG_EP_CTRL,
			"unidentified operation type\n");
	}

	//Skip locking on transfers from self since we already have
	//the ep->region->lock
	if (peer_smr != ep->region) {
		if (fastlock_tryacquire(&peer_smr->lock))
			return -FI_EAGAIN;
	}

	peer_smr->cmd_cnt++;
	if (tx_buf) {
		smr_freestack_push(smr_inject_pool(peer_smr), tx_buf);
	} else if (sar_msg) {
		smr_freestack_push(smr_sar_pool(peer_smr), sar_msg);
		peer_smr->sar_cnt++;
		smr_peer_data(ep->region)[pending->addr].sar_status = 0;
	}

	if (peer_smr != ep->region)
		fastlock_release(&peer_smr->lock);

	return 0;
}

static void smr_progress_resp(struct smr_ep *ep)
{
	struct smr_resp *resp;
	struct smr_tx_entry *pending;
	int ret;

	fastlock_acquire(&ep->region->lock);
	fastlock_acquire(&ep->util_ep.tx_cq->cq_lock);
	while (!ofi_cirque_isempty(smr_resp_queue(ep->region)) &&
	       !ofi_cirque_isfull(ep->util_ep.tx_cq->cirq)) {
		resp = ofi_cirque_head(smr_resp_queue(ep->region));
		if (resp->status == FI_EBUSY)
			break;

		pending = (struct smr_tx_entry *) resp->msg_id;
		if (smr_progress_resp_entry(ep, resp, pending, &resp->status))
			break;

		ret = smr_complete_tx(ep, pending->context,
				  pending->cmd.msg.hdr.op, pending->cmd.msg.hdr.op_flags,
				  -(resp->status));
		if (ret) {
			FI_WARN(&smr_prov, FI_LOG_EP_CTRL,
				"unable to process tx completion\n");
			break;
		}
		freestack_push(ep->pend_fs, pending);
		ofi_cirque_discard(smr_resp_queue(ep->region));
	}
	fastlock_release(&ep->util_ep.tx_cq->cq_lock);
	fastlock_release(&ep->region->lock);
}

static int smr_progress_inline(struct smr_cmd *cmd, struct iovec *iov,
			       size_t iov_count, size_t *total_len)
{
	*total_len = ofi_copy_to_iov(iov, iov_count, 0, cmd->msg.data.msg,
				     cmd->msg.hdr.size);
	if (*total_len != cmd->msg.hdr.size) {
		FI_WARN(&smr_prov, FI_LOG_EP_CTRL,
			"recv truncated");
		return -FI_EIO;
	}
	return 0;
}

static int smr_progress_inject(struct smr_cmd *cmd, struct iovec *iov,
			       size_t iov_count, size_t *total_len,
			       struct smr_ep *ep, int err)
{
	struct smr_inject_buf *tx_buf;
	size_t inj_offset;

	inj_offset = (size_t) cmd->msg.hdr.src_data;
	tx_buf = smr_get_ptr(ep->region, inj_offset);

	if (err) {
		smr_freestack_push(smr_inject_pool(ep->region), tx_buf);
		return err;
	}

	if (cmd->msg.hdr.op == ofi_op_read_req) {
		*total_len = ofi_copy_from_iov(tx_buf->data, cmd->msg.hdr.size,
					       iov, iov_count, 0);
	} else {
		*total_len = ofi_copy_to_iov(iov, iov_count, 0, tx_buf->data,
					     cmd->msg.hdr.size);
		smr_freestack_push(smr_inject_pool(ep->region), tx_buf);
	}

	if (*total_len != cmd->msg.hdr.size) {
		FI_WARN(&smr_prov, FI_LOG_EP_CTRL,
			"recv truncated");
		return -FI_EIO;
	}

	return FI_SUCCESS;
}

static int smr_progress_iov(struct smr_cmd *cmd, struct iovec *iov,
			    size_t iov_count, size_t *total_len,
			    struct smr_ep *ep, int err)
{
	struct smr_region *peer_smr;
	struct smr_resp *resp;
	int peer_id, ret;

	peer_id = (int) cmd->msg.hdr.addr;
	peer_smr = smr_peer_region(ep->region, peer_id);
	resp = smr_get_ptr(peer_smr, cmd->msg.hdr.src_data);

	if (err) {
		ret = -err;
		goto out;
	}

	if (cmd->msg.hdr.op == ofi_op_read_req) {
		ret = ofi_process_vm_writev(peer_smr->pid, iov, iov_count,
					    cmd->msg.data.iov,
					    cmd->msg.data.iov_count, 0);
	} else {
		ret = ofi_process_vm_readv(peer_smr->pid, iov, iov_count,
					   cmd->msg.data.iov,
					   cmd->msg.data.iov_count, 0);
	}

	if (ret != cmd->msg.hdr.size) {
		if (ret < 0) {
			FI_WARN(&smr_prov, FI_LOG_EP_CTRL,
				"CMA write error\n");
			ret = errno;
		} else { 
			FI_WARN(&smr_prov, FI_LOG_EP_CTRL,
				"partial read occurred\n");
			ret = FI_EIO;
		}
	} else {
		*total_len = ret;
		ret = 0;
	}

out:
	//Status must be set last (signals peer: op done, valid resp entry)
	resp->status = ret;

	return -ret;
}

static int smr_mmap_peer_copy(struct smr_ep *ep, struct smr_cmd *cmd,
				 struct iovec *iov, size_t iov_count,
				 size_t *total_len)
{
	char shm_name[NAME_MAX];
	void *mapped_ptr;
	int peer_id, fd, num;
	int ret = 0;

	peer_id = (int) cmd->msg.hdr.addr;

	num = smr_mmap_name(shm_name, ep->region->map->peers[peer_id].peer.name,
			    cmd->msg.hdr.msg_id);
	if (num < 0) {
		FI_WARN(&smr_prov, FI_LOG_AV, "generating shm file name failed\n");
		return -errno;
	}

	fd = shm_open(shm_name, O_RDWR, S_IRUSR | S_IWUSR);
	if (fd < 0) {
		FI_WARN(&smr_prov, FI_LOG_AV, "shm_open error\n");
		return -errno;
	}

	mapped_ptr = mmap(NULL, cmd->msg.hdr.size, PROT_READ | PROT_WRITE,
			  MAP_SHARED, fd, 0);
	if (mapped_ptr == MAP_FAILED) {
		FI_WARN(&smr_prov, FI_LOG_AV, "mmap error %s\n", strerror(errno));
		ret = -errno;
		goto unlink_close;
	}

	if (cmd->msg.hdr.op == ofi_op_read_req) {
		*total_len = ofi_total_iov_len(iov, iov_count);
		if (ofi_copy_from_iov(mapped_ptr, *total_len, iov, iov_count, 0)
		    != *total_len) {
			FI_WARN(&smr_prov, FI_LOG_EP_CTRL,
				"mmap iov copy in error\n");
			ret = -FI_EIO;
			goto munmap;
		}
	} else {
		*total_len = ofi_copy_to_iov(iov, iov_count, 0, mapped_ptr,
				      cmd->msg.hdr.size);
		if (*total_len != cmd->msg.hdr.size) {
			FI_WARN(&smr_prov, FI_LOG_EP_CTRL,
				"mmap iov copy out error\n");
			ret = -FI_EIO;
			goto munmap;
		}
	}

munmap:
	munmap(mapped_ptr, cmd->msg.hdr.size);
unlink_close:
	shm_unlink(shm_name);
	close(fd);
	return ret;
}

static int smr_progress_mmap(struct smr_cmd *cmd, struct iovec *iov,
			     size_t iov_count, size_t *total_len,
			     struct smr_ep *ep)
{
	struct smr_region *peer_smr;
	struct smr_resp *resp;
	int peer_id, ret;

	peer_id = (int) cmd->msg.hdr.addr;
	peer_smr = smr_peer_region(ep->region, peer_id);
	resp = smr_get_ptr(peer_smr, cmd->msg.hdr.src_data);

	ret = smr_mmap_peer_copy(ep, cmd, iov, iov_count, total_len);

	//Status must be set last (signals peer: op done, valid resp entry)
	resp->status = ret;

	return ret;
}

static struct smr_sar_entry *smr_progress_sar(struct smr_cmd *cmd,
			struct smr_rx_entry *rx_entry, struct iovec *iov,
			size_t iov_count, size_t *total_len, struct smr_ep *ep)
{
	struct smr_region *peer_smr;
	struct smr_sar_entry *sar_entry;
	struct smr_sar_msg *sar_msg;
	struct smr_resp *resp;
	struct iovec sar_iov[SMR_IOV_LIMIT];
	int next = 0;

	sar_msg = smr_get_ptr(ep->region, cmd->msg.data.sar);
	peer_smr = smr_peer_region(ep->region, cmd->msg.hdr.addr);
	resp = smr_get_ptr(peer_smr, cmd->msg.hdr.src_data);

	memcpy(sar_iov, iov, sizeof(*iov) * iov_count);
	(void) ofi_truncate_iov(sar_iov, &iov_count, cmd->msg.hdr.size);

	if (cmd->msg.hdr.op == ofi_op_read_req)
		smr_try_progress_to_sar(sar_msg, resp, cmd, sar_iov, iov_count,
					total_len, &next);
	else
		smr_try_progress_from_sar(sar_msg, resp, cmd, sar_iov, iov_count,
					  total_len, &next);

	if (*total_len == cmd->msg.hdr.size)
		return NULL;

	sar_entry = freestack_pop(ep->sar_fs);

	sar_entry->cmd = *cmd;
	sar_entry->bytes_done = *total_len;
	sar_entry->next = next;
	memcpy(sar_entry->iov, sar_iov, sizeof(*sar_iov) * iov_count);
	sar_entry->iov_count = iov_count;
	if (rx_entry) {
		sar_entry->rx_entry = *rx_entry;
		sar_entry->rx_entry.flags |= cmd->msg.hdr.op_flags;
		sar_entry->rx_entry.flags &= ~SMR_MULTI_RECV;
	} else {
		sar_entry->rx_entry.flags = cmd->msg.hdr.op_flags;
	}

	dlist_insert_tail(&sar_entry->entry, &ep->sar_list);
	*total_len = cmd->msg.hdr.size;
	return sar_entry;
}

static bool smr_progress_multi_recv(struct smr_ep *ep,
				    struct smr_rx_entry *entry, size_t len)
{
	size_t left;
	void *new_base;

	left = entry->iov[0].iov_len - len;
	if (left < ep->min_multi_recv_size)
		return true;

	new_base = (void *) ((uintptr_t) entry->iov[0].iov_base + len);
	entry->iov[0].iov_len = left;
	entry->iov[0].iov_base = new_base;

	return false;
}

static void smr_do_atomic(void *src, void *dst, void *cmp, enum fi_datatype datatype,
			  enum fi_op op, size_t cnt, uint16_t flags)
{
	char tmp_result[SMR_INJECT_SIZE];

	if (op >= OFI_SWAP_OP_START) {
		ofi_atomic_swap_handlers[op - OFI_SWAP_OP_START][datatype](dst,
			src, cmp, tmp_result, cnt);
	} else if (flags & SMR_RMA_REQ) {
		ofi_atomic_readwrite_handlers[op][datatype](dst, src,
			tmp_result, cnt);
	} else if (op != FI_ATOMIC_READ) {
		ofi_atomic_write_handlers[op][datatype](dst, src, cnt);
	}

	if (flags & SMR_RMA_REQ)
		memcpy(src, op == FI_ATOMIC_READ ? dst : tmp_result,
		       cnt * ofi_datatype_size(datatype));
}

static int smr_progress_inline_atomic(struct smr_cmd *cmd, struct fi_ioc *ioc,
			       size_t ioc_count, size_t *len)
{
	int i;
	uint8_t *src, *comp;

	switch (cmd->msg.hdr.op) {
	case ofi_op_atomic_compare:
		src = cmd->msg.data.buf;
		comp = cmd->msg.data.comp;
		break;
	default:
		src = cmd->msg.data.msg;
		comp = NULL;
		break;
	}

	for (i = *len = 0; i < ioc_count && *len < cmd->msg.hdr.size; i++) {
		smr_do_atomic(&src[*len], ioc[i].addr, comp ? &comp[*len] : NULL,
			      cmd->msg.hdr.datatype, cmd->msg.hdr.atomic_op,
			      ioc[i].count, cmd->msg.hdr.op_flags);
		*len += ioc[i].count * ofi_datatype_size(cmd->msg.hdr.datatype);
	}

	if (*len != cmd->msg.hdr.size) {
		FI_WARN(&smr_prov, FI_LOG_EP_CTRL,
			"recv truncated");
		return -FI_EIO;
	}
	return 0;
}

static int smr_progress_inject_atomic(struct smr_cmd *cmd, struct fi_ioc *ioc,
			       size_t ioc_count, size_t *len,
			       struct smr_ep *ep, int err)
{
	struct smr_inject_buf *tx_buf;
	size_t inj_offset;
	uint8_t *src, *comp;
	int i;

	inj_offset = (size_t) cmd->msg.hdr.src_data;
	tx_buf = smr_get_ptr(ep->region, inj_offset);
	if (err)
		goto out;

	switch (cmd->msg.hdr.op) {
	case ofi_op_atomic_compare:
		src = tx_buf->buf;
		comp = tx_buf->comp;
		break;
	default:
		src = tx_buf->data;
		comp = NULL;
		break;
	}

	for (i = *len = 0; i < ioc_count && *len < cmd->msg.hdr.size; i++) {
		smr_do_atomic(&src[*len], ioc[i].addr, comp ? &comp[*len] : NULL,
			      cmd->msg.hdr.datatype, cmd->msg.hdr.atomic_op,
			      ioc[i].count, cmd->msg.hdr.op_flags);
		*len += ioc[i].count * ofi_datatype_size(cmd->msg.hdr.datatype);
	}

	if (*len != cmd->msg.hdr.size) {
		FI_WARN(&smr_prov, FI_LOG_EP_CTRL,
			"recv truncated");
		err = -FI_EIO;
	}

out:
	if (!(cmd->msg.hdr.op_flags & SMR_RMA_REQ))
		smr_freestack_push(smr_inject_pool(ep->region), tx_buf);

	return err;
}

static int smr_progress_msg_common(struct smr_ep *ep, struct smr_cmd *cmd,
				   struct smr_rx_entry *entry)
{
	struct smr_sar_entry *sar = NULL;
	size_t total_len = 0;
	uint16_t comp_flags;
	void *comp_buf;
	int ret;
	bool free_entry = true;

	switch (cmd->msg.hdr.op_src) {
	case smr_src_inline:
		entry->err = smr_progress_inline(cmd, entry->iov, entry->iov_count,
						 &total_len);
		ep->region->cmd_cnt++;
		break;
	case smr_src_inject:
		entry->err = smr_progress_inject(cmd, entry->iov, entry->iov_count,
						 &total_len, ep, 0);
		ep->region->cmd_cnt++;
		break;
	case smr_src_iov:
		entry->err = smr_progress_iov(cmd, entry->iov, entry->iov_count,
					      &total_len, ep, 0);
		break;
	case smr_src_mmap:
		entry->err = smr_progress_mmap(cmd, entry->iov, entry->iov_count,
					       &total_len, ep);
		break;
	case smr_src_sar:
		sar = smr_progress_sar(cmd, entry, entry->iov, entry->iov_count,
				       &total_len, ep);
		break;
	default:
		FI_WARN(&smr_prov, FI_LOG_EP_CTRL,
			"unidentified operation type\n");
		entry->err = -FI_EINVAL;
	}

	comp_buf = entry->iov[0].iov_base;
	comp_flags = (cmd->msg.hdr.op_flags | entry->flags) & ~SMR_MULTI_RECV;

	if (entry->flags & SMR_MULTI_RECV) {
		free_entry = smr_progress_multi_recv(ep, entry, total_len);
		if (free_entry) {
			comp_flags |= SMR_MULTI_RECV;
			if (sar)
				sar->rx_entry.flags |= SMR_MULTI_RECV;
		}
	}

	if (!sar) {
		ret = smr_complete_rx(ep, entry->context, cmd->msg.hdr.op,
				comp_flags, total_len, comp_buf, cmd->msg.hdr.addr,
				cmd->msg.hdr.tag, cmd->msg.hdr.data, entry->err);
		if (ret) {
			FI_WARN(&smr_prov, FI_LOG_EP_CTRL,
				"unable to process rx completion\n");
		}
	}

	if (free_entry) {
		dlist_remove(&entry->entry);
		freestack_push(ep->recv_fs, entry);
		return 1;
	}
	return 0;
}

static int smr_progress_cmd_msg(struct smr_ep *ep, struct smr_cmd *cmd)
{
	struct smr_queue *recv_queue;
	struct smr_match_attr match_attr;
	struct dlist_entry *dlist_entry;
	struct smr_unexp_msg *unexp;
	int ret;

	if (ofi_cirque_isfull(ep->util_ep.rx_cq->cirq)) {
		FI_WARN(&smr_prov, FI_LOG_EP_CTRL,
			"rx cq full\n");
		return -FI_ENOSPC;
	}

	recv_queue = (cmd->msg.hdr.op == ofi_op_tagged) ?
		      &ep->trecv_queue : &ep->recv_queue;

	match_attr.addr = cmd->msg.hdr.addr;
	match_attr.tag = cmd->msg.hdr.tag;

	dlist_entry = dlist_find_first_match(&recv_queue->list,
					     recv_queue->match_func,
					     &match_attr);
	if (!dlist_entry) {
		if (freestack_isempty(ep->unexp_fs))
			return -FI_EAGAIN;
		unexp = freestack_pop(ep->unexp_fs);
		memcpy(&unexp->cmd, cmd, sizeof(*cmd));
		ofi_cirque_discard(smr_cmd_queue(ep->region));
		if (cmd->msg.hdr.op == ofi_op_msg) {
			dlist_insert_tail(&unexp->entry, &ep->unexp_msg_queue.list);
		} else {
			assert(cmd->msg.hdr.op == ofi_op_tagged);
			dlist_insert_tail(&unexp->entry, &ep->unexp_tagged_queue.list);
		}
		return 0;
	}
	ret = smr_progress_msg_common(ep, cmd,
			container_of(dlist_entry, struct smr_rx_entry, entry));
	ofi_cirque_discard(smr_cmd_queue(ep->region));
	return ret < 0 ? ret : 0;
}

static int smr_progress_cmd_rma(struct smr_ep *ep, struct smr_cmd *cmd)
{
	struct smr_region *peer_smr;
	struct smr_domain *domain;
	struct smr_cmd *rma_cmd;
	struct smr_resp *resp;
	struct iovec iov[SMR_IOV_LIMIT];
	size_t iov_count;
	size_t total_len = 0;
	int err = 0, ret = 0;

	domain = container_of(ep->util_ep.domain, struct smr_domain,
			      util_domain);

	if (cmd->msg.hdr.op_flags & SMR_REMOTE_CQ_DATA &&
	    ofi_cirque_isfull(ep->util_ep.rx_cq->cirq)) {
		FI_WARN(&smr_prov, FI_LOG_EP_CTRL,
			"rx cq full\n");
		return -FI_ENOSPC;
	}

	ofi_cirque_discard(smr_cmd_queue(ep->region));
	ep->region->cmd_cnt++;
	rma_cmd = ofi_cirque_head(smr_cmd_queue(ep->region));

	for (iov_count = 0; iov_count < rma_cmd->rma.rma_count; iov_count++) {
		ret = ofi_mr_verify(&domain->util_domain.mr_map,
				rma_cmd->rma.rma_iov[iov_count].len,
				(uintptr_t *) &(rma_cmd->rma.rma_iov[iov_count].addr),
				rma_cmd->rma.rma_iov[iov_count].key,
				ofi_rx_mr_reg_flags(cmd->msg.hdr.op, 0));
		if (ret)
			break;

		iov[iov_count].iov_base = (void *) rma_cmd->rma.rma_iov[iov_count].addr;
		iov[iov_count].iov_len = rma_cmd->rma.rma_iov[iov_count].len;
	}
	ofi_cirque_discard(smr_cmd_queue(ep->region));
	if (ret) {
		ep->region->cmd_cnt++;
		return ret;
	}

	switch (cmd->msg.hdr.op_src) {
	case smr_src_inline:
		err = smr_progress_inline(cmd, iov, iov_count, &total_len);
		ep->region->cmd_cnt++;
		break;
	case smr_src_inject:
		err = smr_progress_inject(cmd, iov, iov_count, &total_len, ep, ret);
		if (cmd->msg.hdr.op == ofi_op_read_req && cmd->msg.hdr.data) {
			peer_smr = smr_peer_region(ep->region, cmd->msg.hdr.addr);
			resp = smr_get_ptr(peer_smr, cmd->msg.hdr.data);
			resp->status = -err;
		} else {
			ep->region->cmd_cnt++;
		}
		break;
	case smr_src_iov:
		err = smr_progress_iov(cmd, iov, iov_count, &total_len, ep, ret);
		break;
	case smr_src_mmap:
		err = smr_progress_mmap(cmd, iov, iov_count, &total_len, ep);
		break;
	case smr_src_sar:
		if (smr_progress_sar(cmd, NULL, iov, iov_count, &total_len, ep))
			return ret;
		break;
	default:
		FI_WARN(&smr_prov, FI_LOG_EP_CTRL,
			"unidentified operation type\n");
		err = -FI_EINVAL;
	}

	ret = smr_complete_rx(ep, (void *) cmd->msg.hdr.msg_id,
			      cmd->msg.hdr.op, cmd->msg.hdr.op_flags,
			      total_len, iov_count ? iov[0].iov_base : NULL,
			      cmd->msg.hdr.addr, 0, cmd->msg.hdr.data, err);
	if (ret) {
		FI_WARN(&smr_prov, FI_LOG_EP_CTRL,
		"unable to process rx completion\n");
	}

	return ret;
}

static int smr_progress_cmd_atomic(struct smr_ep *ep, struct smr_cmd *cmd)
{
	struct smr_region *peer_smr;
	struct smr_domain *domain;
	struct smr_cmd *rma_cmd;
	struct smr_resp *resp;
	struct fi_ioc ioc[SMR_IOV_LIMIT];
	size_t ioc_count;
	size_t total_len = 0;
	int err, ret = 0;

	domain = container_of(ep->util_ep.domain, struct smr_domain,
			      util_domain);

	ofi_cirque_discard(smr_cmd_queue(ep->region));
	ep->region->cmd_cnt++;
	rma_cmd = ofi_cirque_head(smr_cmd_queue(ep->region));

	for (ioc_count = 0; ioc_count < rma_cmd->rma.rma_count; ioc_count++) {
		ret = ofi_mr_verify(&domain->util_domain.mr_map,
				rma_cmd->rma.rma_ioc[ioc_count].count *
				ofi_datatype_size(cmd->msg.hdr.datatype),
				(uintptr_t *) &(rma_cmd->rma.rma_ioc[ioc_count].addr),
				rma_cmd->rma.rma_ioc[ioc_count].key,
				ofi_rx_mr_reg_flags(cmd->msg.hdr.op,
				cmd->msg.hdr.atomic_op));
		if (ret)
			break;

		ioc[ioc_count].addr = (void *) rma_cmd->rma.rma_ioc[ioc_count].addr;
		ioc[ioc_count].count = rma_cmd->rma.rma_ioc[ioc_count].count;
	}
	ofi_cirque_discard(smr_cmd_queue(ep->region));
	if (ret) {
		ep->region->cmd_cnt++;
		return ret;
	}

	switch (cmd->msg.hdr.op_src) {
	case smr_src_inline:
		err = smr_progress_inline_atomic(cmd, ioc, ioc_count, &total_len);
		break;
	case smr_src_inject:
		err = smr_progress_inject_atomic(cmd, ioc, ioc_count, &total_len, ep, ret);
		break;
	default:
		FI_WARN(&smr_prov, FI_LOG_EP_CTRL,
			"unidentified operation type\n");
		err = -FI_EINVAL;
	}
	if (cmd->msg.hdr.data) {
		peer_smr = smr_peer_region(ep->region, cmd->msg.hdr.addr);
		resp = smr_get_ptr(peer_smr, cmd->msg.hdr.data);
		resp->status = -err;
	} else {
		ep->region->cmd_cnt++;
	}

	if (err)
		FI_WARN(&smr_prov, FI_LOG_EP_CTRL,
			"error processing atomic op\n");

	ret = smr_complete_rx(ep, NULL, cmd->msg.hdr.op, cmd->msg.hdr.op_flags,
			      total_len, ioc_count ? ioc[0].addr : NULL,
			      cmd->msg.hdr.addr, 0, cmd->msg.hdr.data, err);
	if (ret)
		return ret;

	return err; 
}

static void smr_progress_cmd(struct smr_ep *ep)
{
	struct smr_cmd *cmd;
	int ret = 0;

	fastlock_acquire(&ep->region->lock);
	fastlock_acquire(&ep->util_ep.rx_cq->cq_lock);

	while (!ofi_cirque_isempty(smr_cmd_queue(ep->region))) {
		cmd = ofi_cirque_head(smr_cmd_queue(ep->region));

		switch (cmd->msg.hdr.op) {
		case ofi_op_msg:
		case ofi_op_tagged:
			ret = smr_progress_cmd_msg(ep, cmd);
			break;
		case ofi_op_write:
		case ofi_op_read_req:
			ret = smr_progress_cmd_rma(ep, cmd);
			break;
		case ofi_op_write_async:
		case ofi_op_read_async:
			ofi_ep_rx_cntr_inc_func(&ep->util_ep, cmd->msg.hdr.op);
			ofi_cirque_discard(smr_cmd_queue(ep->region));
			ep->region->cmd_cnt++;
			break;
		case ofi_op_atomic:
		case ofi_op_atomic_fetch:
		case ofi_op_atomic_compare:
			ret = smr_progress_cmd_atomic(ep, cmd);
			break;
		default:
			FI_WARN(&smr_prov, FI_LOG_EP_CTRL,
				"unidentified operation type\n");
			ret = -FI_EINVAL;
		}

		if (ret) {
			if (ret != -FI_EAGAIN) {
				FI_WARN(&smr_prov, FI_LOG_EP_CTRL,
					"error processing command\n");
			}
			break;
		}
	}
	fastlock_release(&ep->util_ep.rx_cq->cq_lock);
	fastlock_release(&ep->region->lock);
}

static void smr_progress_sar_list(struct smr_ep *ep)
{
	struct smr_region *peer_smr;
	struct smr_sar_msg *sar_msg;
	struct smr_sar_entry *sar_entry;
	struct smr_resp *resp;
	struct dlist_entry *tmp;
	int ret;
 
	fastlock_acquire(&ep->region->lock);
	fastlock_acquire(&ep->util_ep.rx_cq->cq_lock);

	dlist_foreach_container_safe(&ep->sar_list, struct smr_sar_entry,
				     sar_entry, entry, tmp) {
		sar_msg = smr_get_ptr(ep->region, sar_entry->cmd.msg.data.sar);
		peer_smr = smr_peer_region(ep->region, sar_entry->cmd.msg.hdr.addr);
		resp = smr_get_ptr(peer_smr, sar_entry->cmd.msg.hdr.src_data);
		if (sar_entry->cmd.msg.hdr.op == ofi_op_read_req)
			smr_try_progress_to_sar(sar_msg, resp, &sar_entry->cmd,
					sar_entry->iov, sar_entry->iov_count,
					&sar_entry->bytes_done, &sar_entry->next);
		else
			smr_try_progress_from_sar(sar_msg, resp, &sar_entry->cmd,
					sar_entry->iov, sar_entry->iov_count,
					&sar_entry->bytes_done, &sar_entry->next);

		if (sar_entry->bytes_done == sar_entry->cmd.msg.hdr.size) {
			ret = smr_complete_rx(ep, sar_entry->rx_entry.context,
					sar_entry->cmd.msg.hdr.op,
					sar_entry->rx_entry.flags,
					sar_entry->bytes_done,
					sar_entry->rx_entry.iov[0].iov_base,
					sar_entry->cmd.msg.hdr.addr,
					sar_entry->cmd.msg.hdr.tag,
					sar_entry->cmd.msg.hdr.data, 0);
			if (ret) {
				FI_WARN(&smr_prov, FI_LOG_EP_CTRL,
					"unable to process rx completion\n");
			}
			dlist_remove(&sar_entry->entry);
			freestack_push(ep->sar_fs, sar_entry);
		}
	}
	fastlock_release(&ep->util_ep.rx_cq->cq_lock);
	fastlock_release(&ep->region->lock);
}

void smr_ep_progress(struct util_ep *util_ep)
{
	struct smr_ep *ep;

	ep = container_of(util_ep, struct smr_ep, util_ep);

	smr_progress_resp(ep);
	smr_progress_cmd(ep);

	if (ep->region->cma_cap == SMR_CMA_CAP_OFF)
		smr_progress_sar_list(ep);
}

int smr_progress_unexp_queue(struct smr_ep *ep, struct smr_rx_entry *entry,
			     struct smr_queue *unexp_queue)
{
	struct smr_match_attr match_attr;
	struct smr_unexp_msg *unexp_msg;
	struct dlist_entry *dlist_entry;
	int multi_recv;
	int ret;

	match_attr.addr = entry->addr;
	match_attr.ignore = entry->ignore;
	match_attr.tag = entry->tag;

	dlist_entry = dlist_remove_first_match(&unexp_queue->list,
					       unexp_queue->match_func,
					       &match_attr);
	if (!dlist_entry)
		return 0;

	multi_recv = entry->flags & SMR_MULTI_RECV;
	while (dlist_entry) {
		unexp_msg = container_of(dlist_entry, struct smr_unexp_msg, entry);
		ret = smr_progress_msg_common(ep, &unexp_msg->cmd, entry);
		freestack_push(ep->unexp_fs, unexp_msg);
		if (!multi_recv || ret)
			break;

		dlist_entry = dlist_remove_first_match(&unexp_queue->list,
						       unexp_queue->match_func,
						       &match_attr);
	}

	return ret < 0 ? ret : 0;
}