include/ccKVS/inline_util.h
static inline void poll_and_send_remotes(uint16_t previous_wr_i, uint16_t local_client_id, uint32_t* outstanding_rem_reqs,
										 struct hrd_ctrl_blk* cb, struct ibv_wc* wc, uint16_t prev_rem_req_i, uint16_t wr_i,
										 struct ibv_send_wr* rem_send_wr, uint16_t rem_req_i, long long* remote_tot_tx,
										 struct ibv_sge* rem_send_sgl, struct latency_flags* latency_info, struct timespec* start)
{
	int ret;
	struct ibv_send_wr* bad_send_wr;
	uint16_t i,j;
	//Poll for remote request receive completions
	if (ENABLE_MULTI_BATCHES == 0) {
		if(previous_wr_i > 0) {
			//			if (local_client_id == 0) printf("polling remote recv compeltions %d \n", prev_rem_req_i);
			c_stats[local_client_id].stalled_time_per_client +=
					hrd_poll_cq(cb->dgram_recv_cq[REMOTE_UD_QP_ID], prev_rem_req_i, wc);
			// if (local_client_id == 0) printf("polled %d\n", prev_rem_req_i );
			*outstanding_rem_reqs -= previous_wr_i;
			if (ENABLE_ASSERTIONS) assert(prev_rem_req_i <= MAX_REMOTE_RECV_WCS);
			if ((MEASURE_LATENCY == 1) && ((latency_info->measured_req_flag) == REMOTE_REQ)) {
				report_remote_latency(latency_info, prev_rem_req_i, wc, start);
			}
		}
	}

	// Send remote requests
	if (wr_i > 0) {
		if (ENABLE_ASSERTIONS) {
			if (rem_req_i > 0) assert(wr_i > 0);
			assert(rem_req_i <= wr_i * MAX_COALESCE_PER_MACH);
			assert(wr_i <= WORKERS_PER_MACHINE * WS_PER_WORKER * (MACHINE_NUM- 1));
			assert(wr_i <= WINDOW_SIZE);
		}
		if (DEBUG_COALESCING && ENABLE_COALESCING)
			debug_coalescing(rem_send_sgl, local_client_id, rem_req_i, wr_i);
			//		yellow_printf("Sending %d messages \n", wr_i);
		rem_send_wr[wr_i - 1].next = NULL;
		ret = ibv_post_send(cb->dgram_qp[REMOTE_UD_QP_ID], &rem_send_wr[0], &bad_send_wr);
		if (ENABLE_ASSERTIONS && ret != 0)
			printf("Batch size: %d, already outstanding %d  \n", wr_i, *outstanding_rem_reqs);
		CPE(ret, "Sending Remotes: ibv_post_send error", ret);
		// if (MEASURE_LATENCY == 1) rem_send_wr[last_measured_wr_i].imm_data = clt_gid;
		c_stats[local_client_id].remote_messages_per_client += wr_i;
		//		if (ENABLE_WORKER_COALESCING == 0)
		//			c_stats[local_client_id].remotes_per_client += wr_i;
		*outstanding_rem_reqs += wr_i;
		if (ENABLE_ASSERTIONS == 1) {
			if (*outstanding_rem_reqs > MAX_OUTSTANDING_REQS) {
				yellow_printf("Clt %d, wr_i: %d, outstanding_rem_reqs %d \n",
							  local_client_id, wr_i,	*outstanding_rem_reqs);
				assert(0);
			}
		}
		(*remote_tot_tx) += wr_i; // used for Selective Signaling -- DONT CHANGE!
		c_stats[local_client_id].batches_per_client++;

	}
	else if (ENABLE_STAT_COUNTING == 1) c_stats[local_client_id].wasted_loops++;
}
/* Poll for receive compeltions for answers to remote requests and then
	 send the new batch of remote requests to the NIC */
Poll for receive compeltions for answers to remote requests and then
send the new batch of remote requests to the NIC
Poll for receive compeltions for answers to remote requests and then send the new batch of remote requests to the NIC
static inline void poll_and_send_remotes(uint16_t previous_wr_i, uint16_t local_client_id, uint32_t* outstanding_rem_reqs,
										 struct hrd_ctrl_blk* cb, struct ibv_wc* wc, uint16_t prev_rem_req_i, uint16_t wr_i,
										 struct ibv_send_wr* rem_send_wr, uint16_t rem_req_i, long long* remote_tot_tx,
										 struct ibv_sge* rem_send_sgl, struct latency_flags* latency_info, struct timespec* start)
{
	int ret;
	struct ibv_send_wr* bad_send_wr;
	uint16_t i,j;
	if (ENABLE_MULTI_BATCHES == 0) {
		if(previous_wr_i > 0) {
			c_stats[local_client_id].stalled_time_per_client +=
					hrd_poll_cq(cb->dgram_recv_cq[REMOTE_UD_QP_ID], prev_rem_req_i, wc);
			*outstanding_rem_reqs -= previous_wr_i;
			if (ENABLE_ASSERTIONS) assert(prev_rem_req_i <= MAX_REMOTE_RECV_WCS);
			if ((MEASURE_LATENCY == 1) && ((latency_info->measured_req_flag) == REMOTE_REQ)) {
				report_remote_latency(latency_info, prev_rem_req_i, wc, start);
			}
		}
	}
	if (wr_i > 0) {
		if (ENABLE_ASSERTIONS) {
			if (rem_req_i > 0) assert(wr_i > 0);
			assert(rem_req_i <= wr_i * MAX_COALESCE_PER_MACH);
			assert(wr_i <= WORKERS_PER_MACHINE * WS_PER_WORKER * (MACHINE_NUM- 1));
			assert(wr_i <= WINDOW_SIZE);
		}
		if (DEBUG_COALESCING && ENABLE_COALESCING)
			debug_coalescing(rem_send_sgl, local_client_id, rem_req_i, wr_i);
		rem_send_wr[wr_i - 1].next = NULL;
		ret = ibv_post_send(cb->dgram_qp[REMOTE_UD_QP_ID], &rem_send_wr[0], &bad_send_wr);
		if (ENABLE_ASSERTIONS && ret != 0)
			printf("Batch size: %d, already outstanding %d  \n", wr_i, *outstanding_rem_reqs);
		CPE(ret, "Sending Remotes: ibv_post_send error", ret);
		c_stats[local_client_id].remote_messages_per_client += wr_i;
		*outstanding_rem_reqs += wr_i;
		if (ENABLE_ASSERTIONS == 1) {
			if (*outstanding_rem_reqs > MAX_OUTSTANDING_REQS) {
				yellow_printf("Clt %d, wr_i: %d, outstanding_rem_reqs %d \n",
							  local_client_id, wr_i,	*outstanding_rem_reqs);
				assert(0);
			}
		}
		(*remote_tot_tx) += wr_i; 
		c_stats[local_client_id].batches_per_client++;
	}
	else if (ENABLE_STAT_COUNTING == 1) c_stats[local_client_id].wasted_loops++;
}
static inline void poll_and_send_remotes ( uint16_t previous_wr_i , uint16_t local_client_id , uint32_t * outstanding_rem_reqs , struct hrd_ctrl_blk * cb , struct ibv_wc * wc , uint16_t prev_rem_req_i , uint16_t wr_i , struct ibv_send_wr * rem_send_wr , uint16_t rem_req_i , long long * remote_tot_tx , struct ibv_sge * rem_send_sgl , struct latency_flags * latency_info , struct timespec * start ) { int ret ; struct ibv_send_wr * bad_send_wr ; uint16_t i , j ; if ( ENABLE_MULTI_BATCHES == 0 ) { if ( previous_wr_i > 0 ) { c_stats [ local_client_id ] . stalled_time_per_client += hrd_poll_cq ( cb -> dgram_recv_cq [ REMOTE_UD_QP_ID ] , prev_rem_req_i , wc ) ; * outstanding_rem_reqs -= previous_wr_i ; if ( ENABLE_ASSERTIONS ) assert ( prev_rem_req_i <= MAX_REMOTE_RECV_WCS ) ; if ( ( MEASURE_LATENCY == 1 ) && ( ( latency_info -> measured_req_flag ) == REMOTE_REQ ) ) { report_remote_latency ( latency_info , prev_rem_req_i , wc , start ) ; } } } if ( wr_i > 0 ) { if ( ENABLE_ASSERTIONS ) { if ( rem_req_i > 0 ) assert ( wr_i > 0 ) ; assert ( rem_req_i <= wr_i * MAX_COALESCE_PER_MACH ) ; assert ( wr_i <= WORKERS_PER_MACHINE * WS_PER_WORKER * ( MACHINE_NUM - 1 ) ) ; assert ( wr_i <= WINDOW_SIZE ) ; } if ( DEBUG_COALESCING && ENABLE_COALESCING ) debug_coalescing ( rem_send_sgl , local_client_id , rem_req_i , wr_i ) ; rem_send_wr [ wr_i - 1 ] . next = NULL ; ret = ibv_post_send ( cb -> dgram_qp [ REMOTE_UD_QP_ID ] , & rem_send_wr [ 0 ] , & bad_send_wr ) ; if ( ENABLE_ASSERTIONS && ret != 0 ) printf ( " \n " , wr_i , * outstanding_rem_reqs ) ; CPE ( ret , " " , ret ) ; c_stats [ local_client_id ] . remote_messages_per_client += wr_i ; * outstanding_rem_reqs += wr_i ; if ( ENABLE_ASSERTIONS == 1 ) { if ( * outstanding_rem_reqs > MAX_OUTSTANDING_REQS ) { yellow_printf ( " \n " , local_client_id , wr_i , * outstanding_rem_reqs ) ; assert ( 0 ) ; } } ( * remote_tot_tx ) += wr_i ; c_stats [ local_client_id ] . batches_per_client ++ ; } else if ( ENABLE_STAT_COUNTING == 1 ) c_stats [ local_client_id ] . wasted_loops ++ ; }
Poll for receive compeltions for answers to remote requests and then
send the new batch of remote requests to the NIC
Poll for receive compeltions for answers to remote requests and then send the new batch of remote requests to the NIC
//Poll for remote request receive completions //			if (local_client_id == 0) printf("polling remote recv compeltions %d \n", prev_rem_req_i); // if (local_client_id == 0) printf("polled %d\n", prev_rem_req_i ); // Send remote requests //		yellow_printf("Sending %d messages \n", wr_i); // if (MEASURE_LATENCY == 1) rem_send_wr[last_measured_wr_i].imm_data = clt_gid; //		if (ENABLE_WORKER_COALESCING == 0) //			c_stats[local_client_id].remotes_per_client += wr_i; // used for Selective Signaling -- DONT CHANGE!