include/ccKVS/inline_util.h
static inline void worker_post_receives_and_sends(uint16_t nb_new_req_tot, struct hrd_ctrl_blk* cb, uint16_t* per_qp_received_messages,
												  struct ibv_recv_wr* recv_wr, struct ibv_sge* recv_sgl, int* push_ptr, uint32_t* qp_buf_base,
												  uint32_t* per_qp_buf_slots, uint32_t* debug_recv, uint16_t* clts_per_qp, struct ibv_send_wr* wr,
												  uint16_t send_qp_i, uint16_t wrkr_lid, uint16_t last_measured_wr_i, uint16_t total_reqs_sent)
{
	uint16_t qp_i, i;
	int ret;
	struct ibv_recv_wr *bad_recv_wr;
	struct ibv_send_wr *bad_send_wr;
	// Refill the depleted RECVs before sending anything
	for (qp_i = 0; qp_i < WORKER_NUM_UD_QPS; qp_i++) {
		//      printf("Received %d\n",  per_qp_received_messages[qp_i]);
		// printf("Total slots %d\n", per_qp_buf_slots[qp_i]);
		if(WORKER_NUM_UD_QPS == 1 && nb_new_req_tot > 0 && ENABLE_ASSERTIONS)
			assert(per_qp_received_messages[qp_i] > 0 && per_qp_received_messages[qp_i] <= WORKER_MAX_BATCH);
		for (i = 0; i < per_qp_received_messages[qp_i]; i++) {
			// printf("Receiving at slot  %d\n", qp_buf_base[qp_i] + push_ptr[qp_i] );
			recv_sgl[i].addr = (uintptr_t) &cb->dgram_buf[(qp_buf_base[qp_i] + push_ptr[qp_i]) * sizeof(struct wrkr_ud_req)];
			MOD_ADD_WITH_BASE(push_ptr[qp_i], per_qp_buf_slots[qp_i], 0);
			recv_wr[i].next = (i == per_qp_received_messages[qp_i] - 1) ? NULL : &recv_wr[i + 1];
			if (DEBUG_WORKER_RECVS) (*debug_recv)++;
		}
		if (per_qp_received_messages[qp_i] > 0) {
			ret = ibv_post_recv(cb->dgram_qp[qp_i], &recv_wr[0], &bad_recv_wr);
			CPE(ret, " Worker ibv_post_recv error", ret);
		}
		if (ENABLE_ASSERTIONS && DEBUG_WORKER_RECVS) assert(*debug_recv >= clts_per_qp[0]  * WS_PER_WORKER);
	}

	// Send the batch to the NIC
	if (nb_new_req_tot > 0) {
		wr[nb_new_req_tot - 1].next = NULL;
		ret = ibv_post_send(cb->dgram_qp[send_qp_i], &wr[0], &bad_send_wr);
		CPE(ret, "worker ibv_post_send error", ret);
		//		green_printf("Sending %d responses through qp %d\n", nb_new_req_tot, send_qp_i);
		w_stats[wrkr_lid].batches_per_worker++;
		w_stats[wrkr_lid].remotes_per_worker+= total_reqs_sent;
		if (MEASURE_LATENCY == 1) {
			wr[last_measured_wr_i].opcode = IBV_WR_SEND;
		}
		MOD_ADD_WITH_BASE(send_qp_i, WORKER_NUM_UD_QPS, 0);
	}
}
// Worker post receives on all of its qps and then posts its sends
Worker post receives on all of its qps and then posts its sends
Worker post receives on all of its qps and then posts its sends
static inline void worker_post_receives_and_sends(uint16_t nb_new_req_tot, struct hrd_ctrl_blk* cb, uint16_t* per_qp_received_messages,
												  struct ibv_recv_wr* recv_wr, struct ibv_sge* recv_sgl, int* push_ptr, uint32_t* qp_buf_base,
												  uint32_t* per_qp_buf_slots, uint32_t* debug_recv, uint16_t* clts_per_qp, struct ibv_send_wr* wr,
												  uint16_t send_qp_i, uint16_t wrkr_lid, uint16_t last_measured_wr_i, uint16_t total_reqs_sent)
{
	uint16_t qp_i, i;
	int ret;
	struct ibv_recv_wr *bad_recv_wr;
	struct ibv_send_wr *bad_send_wr;
	for (qp_i = 0; qp_i < WORKER_NUM_UD_QPS; qp_i++) {
		if(WORKER_NUM_UD_QPS == 1 && nb_new_req_tot > 0 && ENABLE_ASSERTIONS)
			assert(per_qp_received_messages[qp_i] > 0 && per_qp_received_messages[qp_i] <= WORKER_MAX_BATCH);
		for (i = 0; i < per_qp_received_messages[qp_i]; i++) {
			recv_sgl[i].addr = (uintptr_t) &cb->dgram_buf[(qp_buf_base[qp_i] + push_ptr[qp_i]) * sizeof(struct wrkr_ud_req)];
			MOD_ADD_WITH_BASE(push_ptr[qp_i], per_qp_buf_slots[qp_i], 0);
			recv_wr[i].next = (i == per_qp_received_messages[qp_i] - 1) ? NULL : &recv_wr[i + 1];
			if (DEBUG_WORKER_RECVS) (*debug_recv)++;
		}
		if (per_qp_received_messages[qp_i] > 0) {
			ret = ibv_post_recv(cb->dgram_qp[qp_i], &recv_wr[0], &bad_recv_wr);
			CPE(ret, " Worker ibv_post_recv error", ret);
		}
		if (ENABLE_ASSERTIONS && DEBUG_WORKER_RECVS) assert(*debug_recv >= clts_per_qp[0]  * WS_PER_WORKER);
	}
	if (nb_new_req_tot > 0) {
		wr[nb_new_req_tot - 1].next = NULL;
		ret = ibv_post_send(cb->dgram_qp[send_qp_i], &wr[0], &bad_send_wr);
		CPE(ret, "worker ibv_post_send error", ret);
		w_stats[wrkr_lid].batches_per_worker++;
		w_stats[wrkr_lid].remotes_per_worker+= total_reqs_sent;
		if (MEASURE_LATENCY == 1) {
			wr[last_measured_wr_i].opcode = IBV_WR_SEND;
		}
		MOD_ADD_WITH_BASE(send_qp_i, WORKER_NUM_UD_QPS, 0);
	}
}
static inline void worker_post_receives_and_sends ( uint16_t nb_new_req_tot , struct hrd_ctrl_blk * cb , uint16_t * per_qp_received_messages , struct ibv_recv_wr * recv_wr , struct ibv_sge * recv_sgl , int * push_ptr , uint32_t * qp_buf_base , uint32_t * per_qp_buf_slots , uint32_t * debug_recv , uint16_t * clts_per_qp , struct ibv_send_wr * wr , uint16_t send_qp_i , uint16_t wrkr_lid , uint16_t last_measured_wr_i , uint16_t total_reqs_sent ) { uint16_t qp_i , i ; int ret ; struct ibv_recv_wr * bad_recv_wr ; struct ibv_send_wr * bad_send_wr ; for ( qp_i = 0 ; qp_i < WORKER_NUM_UD_QPS ; qp_i ++ ) { if ( WORKER_NUM_UD_QPS == 1 && nb_new_req_tot > 0 && ENABLE_ASSERTIONS ) assert ( per_qp_received_messages [ qp_i ] > 0 && per_qp_received_messages [ qp_i ] <= WORKER_MAX_BATCH ) ; for ( i = 0 ; i < per_qp_received_messages [ qp_i ] ; i ++ ) { recv_sgl [ i ] . addr = ( uintptr_t ) & cb -> dgram_buf [ ( qp_buf_base [ qp_i ] + push_ptr [ qp_i ] ) * sizeof ( struct wrkr_ud_req ) ] ; MOD_ADD_WITH_BASE ( push_ptr [ qp_i ] , per_qp_buf_slots [ qp_i ] , 0 ) ; recv_wr [ i ] . next = ( i == per_qp_received_messages [ qp_i ] - 1 ) ? NULL : & recv_wr [ i + 1 ] ; if ( DEBUG_WORKER_RECVS ) ( * debug_recv ) ++ ; } if ( per_qp_received_messages [ qp_i ] > 0 ) { ret = ibv_post_recv ( cb -> dgram_qp [ qp_i ] , & recv_wr [ 0 ] , & bad_recv_wr ) ; CPE ( ret , " " , ret ) ; } if ( ENABLE_ASSERTIONS && DEBUG_WORKER_RECVS ) assert ( * debug_recv >= clts_per_qp [ 0 ] * WS_PER_WORKER ) ; } if ( nb_new_req_tot > 0 ) { wr [ nb_new_req_tot - 1 ] . next = NULL ; ret = ibv_post_send ( cb -> dgram_qp [ send_qp_i ] , & wr [ 0 ] , & bad_send_wr ) ; CPE ( ret , " " , ret ) ; w_stats [ wrkr_lid ] . batches_per_worker ++ ; w_stats [ wrkr_lid ] . remotes_per_worker += total_reqs_sent ; if ( MEASURE_LATENCY == 1 ) { wr [ last_measured_wr_i ] . opcode = IBV_WR_SEND ; } MOD_ADD_WITH_BASE ( send_qp_i , WORKER_NUM_UD_QPS , 0 ) ; } }
Worker post receives on all of its qps and then posts its sends
Worker post receives on all of its qps and then posts its sends
// Refill the depleted RECVs before sending anything //      printf("Received %d\n",  per_qp_received_messages[qp_i]); // printf("Total slots %d\n", per_qp_buf_slots[qp_i]); // printf("Receiving at slot  %d\n", qp_buf_base[qp_i] + push_ptr[qp_i] ); // Send the batch to the NIC //		green_printf("Sending %d responses through qp %d\n", nb_new_req_tot, send_qp_i);