void rxr_pkt_handle_recv_completion()

in prov/efa/src/rxr/rxr_pkt_cmd.c [489:613]

• 112 lines of code
• 34 McCabe index (conditional complexity)

void rxr_pkt_handle_recv_completion(struct rxr_ep *ep,
				    struct fi_cq_data_entry *cq_entry,
				    fi_addr_t src_addr)
{
	struct rxr_peer *peer;
	struct rxr_base_hdr *base_hdr;
	struct rxr_pkt_entry *pkt_entry;

	pkt_entry = (struct rxr_pkt_entry *)cq_entry->op_context;
	pkt_entry->pkt_size = cq_entry->len;
	assert(pkt_entry->pkt_size > 0);

	base_hdr = rxr_get_base_hdr(pkt_entry->pkt);
	if (base_hdr->type >= RXR_EXTRA_REQ_PKT_END) {
		FI_WARN(&rxr_prov, FI_LOG_CQ,
			"Peer %d is requesting feature %d, which this EP does not support.\n",
			(int)src_addr, base_hdr->type);

		assert(0 && "invalid REQ packe type");
		rxr_cq_handle_cq_error(ep, -FI_EIO);
		return;
	}

	if (base_hdr->type >= RXR_REQ_PKT_BEGIN) {
		/*
		 * as long as the REQ packet contain raw address
		 * we will need to call insert because it might be a new
		 * EP with new Q-Key.
		 */
		void *raw_addr;

		raw_addr = rxr_pkt_req_raw_addr(pkt_entry);
		if (OFI_UNLIKELY(raw_addr != NULL))
			pkt_entry->addr = rxr_pkt_insert_addr(ep, pkt_entry, raw_addr);
		else
			pkt_entry->addr = src_addr;
	} else {
		assert(src_addr != FI_ADDR_NOTAVAIL);
		pkt_entry->addr = src_addr;
	}

#if ENABLE_DEBUG
	dlist_remove(&pkt_entry->dbg_entry);
	dlist_insert_tail(&pkt_entry->dbg_entry, &ep->rx_pkt_list);
#ifdef ENABLE_RXR_PKT_DUMP
	rxr_ep_print_pkt("Received", ep, (struct rxr_base_hdr *)pkt_entry->pkt);
#endif
#endif
	peer = rxr_ep_get_peer(ep, pkt_entry->addr);
	if (!(peer->flags & RXR_PEER_HANDSHAKE_SENT))
		rxr_pkt_post_handshake(ep, peer, pkt_entry->addr);

	if (peer->is_local) {
		assert(ep->use_shm);
		ep->posted_bufs_shm--;
	} else {
		ep->posted_bufs_efa--;
	}

	switch (base_hdr->type) {
	case RXR_RETIRED_RTS_PKT:
		FI_WARN(&rxr_prov, FI_LOG_CQ,
			"Received a RTS packet, which has been retired since protocol version 4\n");
		assert(0 && "deprecated RTS pakcet received");
		rxr_cq_handle_cq_error(ep, -FI_EIO);
		return;
	case RXR_RETIRED_CONNACK_PKT:
		FI_WARN(&rxr_prov, FI_LOG_CQ,
			"Received a CONNACK packet, which has been retired since protocol version 4\n");
		assert(0 && "deprecated CONNACK pakcet received");
		rxr_cq_handle_cq_error(ep, -FI_EIO);
		return;
	case RXR_EOR_PKT:
		rxr_pkt_handle_eor_recv(ep, pkt_entry);
		return;
	case RXR_HANDSHAKE_PKT:
		rxr_pkt_handle_handshake_recv(ep, pkt_entry);
		return;
	case RXR_CTS_PKT:
		rxr_pkt_handle_cts_recv(ep, pkt_entry);
		return;
	case RXR_DATA_PKT:
		rxr_pkt_handle_data_recv(ep, pkt_entry);
		return;
	case RXR_READRSP_PKT:
		rxr_pkt_handle_readrsp_recv(ep, pkt_entry);
		return;
	case RXR_ATOMRSP_PKT:
		rxr_pkt_handle_atomrsp_recv(ep, pkt_entry);
		return;
	case RXR_EAGER_MSGRTM_PKT:
	case RXR_EAGER_TAGRTM_PKT:
	case RXR_MEDIUM_MSGRTM_PKT:
	case RXR_MEDIUM_TAGRTM_PKT:
	case RXR_LONG_MSGRTM_PKT:
	case RXR_LONG_TAGRTM_PKT:
	case RXR_READ_MSGRTM_PKT:
	case RXR_READ_TAGRTM_PKT:
	case RXR_WRITE_RTA_PKT:
	case RXR_FETCH_RTA_PKT:
	case RXR_COMPARE_RTA_PKT:
		rxr_pkt_handle_rtm_rta_recv(ep, pkt_entry);
		return;
	case RXR_EAGER_RTW_PKT:
		rxr_pkt_handle_eager_rtw_recv(ep, pkt_entry);
		return;
	case RXR_LONG_RTW_PKT:
		rxr_pkt_handle_long_rtw_recv(ep, pkt_entry);
		return;
	case RXR_READ_RTW_PKT:
		rxr_pkt_handle_read_rtw_recv(ep, pkt_entry);
		return;
	case RXR_SHORT_RTR_PKT:
	case RXR_LONG_RTR_PKT:
		rxr_pkt_handle_rtr_recv(ep, pkt_entry);
		return;
	default:
		FI_WARN(&rxr_prov, FI_LOG_CQ,
			"invalid control pkt type %d\n",
			rxr_get_base_hdr(pkt_entry->pkt)->type);
		assert(0 && "invalid control pkt type");
		rxr_cq_handle_cq_error(ep, -FI_EIO);
		return;
	}
}