include/ccKVS/inline_util.h
static inline void send_acks(struct small_cache_op* inv_to_send_ops, uint8_t credits[][MACHINE_NUM], struct ibv_wc* credit_wc,
							 struct ibv_send_wr* ack_wr, struct ibv_sge* ack_sgl, long long* sent_ack_tx, uint16_t* inv_size,
							 uint16_t *ack_recv_counter, struct ibv_recv_wr* credit_recv_wr, uint16_t local_client_id,
							 uint16_t coh_message_count[][MACHINE_NUM], struct hrd_ctrl_blk* cb, uint16_t debug_ptr)
{
	uint16_t i, send_ack_count = 0, remote_clt_id;
	struct ibv_wc signal_send_wc;
	for (i = 0; i < BCAST_TO_CACHE_BATCH; i++) {
		if (inv_to_send_ops[i].opcode != CACHE_OP_BRC) {
			continue;
		}
		if (credits[ACK_VC][inv_to_send_ops[i].key.meta.cid] == 0)
			poll_credits_LIN(cb->dgram_recv_cq[FC_UD_QP_ID], credit_wc, credits);
		if (credits[ACK_VC][inv_to_send_ops[i].key.meta.cid] == 0) continue;
		//      green_printf(" Sending an ack for key %d with version %d \n",
		//			   inv_to_send_ops[i].key.tag, inv_to_send_ops[i].key.meta.version);
		credits[ACK_VC][inv_to_send_ops[i].key.meta.cid]--;
		remote_clt_id = inv_to_send_ops[i].key.meta.cid * CLIENTS_PER_MACHINE + local_client_id;
		coh_message_count[INV_VC][inv_to_send_ops[i].key.meta.cid]++; // we are emptying that buffer space
		//		printf("inv %d/%d counter %d,  cid: %d, inv_size %d, credits %d, debug_ptr %d \n", i, BCAST_TO_CACHE_BATCH,
		//					 coh_message_count[INV_VC][inv_to_send_ops[i].key.meta.cid], inv_to_send_ops[i].key.meta.cid,
		//					 (*inv_size), credits[ACK_VC][inv_to_send_ops[i].key.meta.cid], debug_ptr);

		ack_wr[send_ack_count].wr.ud.ah = remote_clt_qp[remote_clt_id][BROADCAST_UD_QP_ID].ah;
		ack_wr[send_ack_count].wr.ud.remote_qpn = remote_clt_qp[remote_clt_id][BROADCAST_UD_QP_ID].qpn;
		inv_to_send_ops[i].opcode = CACHE_OP_ACK;
		inv_to_send_ops[i].key.meta.cid = machine_id;
		ack_sgl[send_ack_count].addr = (uint64_t) (uintptr_t) (inv_to_send_ops + i);
		if (send_ack_count > 0) ack_wr[send_ack_count - 1].next = &ack_wr[send_ack_count];
		ack_wr[send_ack_count].send_flags = IBV_SEND_INLINE;
		if ((*sent_ack_tx) % ACK_SS_BATCH == 0) {
			ack_wr[send_ack_count].send_flags |= IBV_SEND_SIGNALED;
			// if (local_client_id == 0) green_printf("Sending ack %llu signaled \n", *sent_ack_tx);
		}
		if((*sent_ack_tx) % ACK_SS_BATCH == ACK_SS_BATCH - 1) {
			// if (local_client_id == 0) green_printf("Polling for ack  %llu \n", *sent_ack_tx);
			hrd_poll_cq(cb->dgram_send_cq[BROADCAST_UD_QP_ID], 1, &signal_send_wc);
		}
		send_ack_count++;
		(*inv_size)--;

		(*sent_ack_tx)++; // Selective signaling
		c_stats[local_client_id].acks_per_client++;
		// Post a receive for a credit every CREDITS_IN_MESSAGE acks
		if ((*sent_ack_tx) % CREDITS_IN_MESSAGE == 0) (*ack_recv_counter)++;
	}
	// post receives for credits and then send the acks
	post_recvs_and_send(*ack_recv_counter, send_ack_count, ack_wr, credit_recv_wr,
						cb->dgram_qp[BROADCAST_UD_QP_ID],cb->dgram_qp[FC_UD_QP_ID], MAX_CREDIT_RECVS,
						BCAST_TO_CACHE_BATCH);
	if (*ack_recv_counter > 0) *ack_recv_counter = 0;
}
// Pick up the buffer with invalidations that succeeded and respond to them with acks
Pick up the buffer with invalidations that succeeded and respond to them with acks
Pick up the buffer with invalidations that succeeded and respond to them with acks
static inline void send_acks(struct small_cache_op* inv_to_send_ops, uint8_t credits[][MACHINE_NUM], struct ibv_wc* credit_wc,
							 struct ibv_send_wr* ack_wr, struct ibv_sge* ack_sgl, long long* sent_ack_tx, uint16_t* inv_size,
							 uint16_t *ack_recv_counter, struct ibv_recv_wr* credit_recv_wr, uint16_t local_client_id,
							 uint16_t coh_message_count[][MACHINE_NUM], struct hrd_ctrl_blk* cb, uint16_t debug_ptr)
{
	uint16_t i, send_ack_count = 0, remote_clt_id;
	struct ibv_wc signal_send_wc;
	for (i = 0; i < BCAST_TO_CACHE_BATCH; i++) {
		if (inv_to_send_ops[i].opcode != CACHE_OP_BRC) {
			continue;
		}
		if (credits[ACK_VC][inv_to_send_ops[i].key.meta.cid] == 0)
			poll_credits_LIN(cb->dgram_recv_cq[FC_UD_QP_ID], credit_wc, credits);
		if (credits[ACK_VC][inv_to_send_ops[i].key.meta.cid] == 0) continue;
		credits[ACK_VC][inv_to_send_ops[i].key.meta.cid]--;
		remote_clt_id = inv_to_send_ops[i].key.meta.cid * CLIENTS_PER_MACHINE + local_client_id;
		coh_message_count[INV_VC][inv_to_send_ops[i].key.meta.cid]++; 
		ack_wr[send_ack_count].wr.ud.ah = remote_clt_qp[remote_clt_id][BROADCAST_UD_QP_ID].ah;
		ack_wr[send_ack_count].wr.ud.remote_qpn = remote_clt_qp[remote_clt_id][BROADCAST_UD_QP_ID].qpn;
		inv_to_send_ops[i].opcode = CACHE_OP_ACK;
		inv_to_send_ops[i].key.meta.cid = machine_id;
		ack_sgl[send_ack_count].addr = (uint64_t) (uintptr_t) (inv_to_send_ops + i);
		if (send_ack_count > 0) ack_wr[send_ack_count - 1].next = &ack_wr[send_ack_count];
		ack_wr[send_ack_count].send_flags = IBV_SEND_INLINE;
		if ((*sent_ack_tx) % ACK_SS_BATCH == 0) {
			ack_wr[send_ack_count].send_flags |= IBV_SEND_SIGNALED;
		}
		if((*sent_ack_tx) % ACK_SS_BATCH == ACK_SS_BATCH - 1) {
			hrd_poll_cq(cb->dgram_send_cq[BROADCAST_UD_QP_ID], 1, &signal_send_wc);
		}
		send_ack_count++;
		(*inv_size)--;
		(*sent_ack_tx)++; 
		c_stats[local_client_id].acks_per_client++;
		if ((*sent_ack_tx) % CREDITS_IN_MESSAGE == 0) (*ack_recv_counter)++;
	}
	post_recvs_and_send(*ack_recv_counter, send_ack_count, ack_wr, credit_recv_wr,
						cb->dgram_qp[BROADCAST_UD_QP_ID],cb->dgram_qp[FC_UD_QP_ID], MAX_CREDIT_RECVS,
						BCAST_TO_CACHE_BATCH);
	if (*ack_recv_counter > 0) *ack_recv_counter = 0;
}
static inline void send_acks ( struct small_cache_op * inv_to_send_ops , uint8_t credits [ ] [ MACHINE_NUM ] , struct ibv_wc * credit_wc , struct ibv_send_wr * ack_wr , struct ibv_sge * ack_sgl , long long * sent_ack_tx , uint16_t * inv_size , uint16_t * ack_recv_counter , struct ibv_recv_wr * credit_recv_wr , uint16_t local_client_id , uint16_t coh_message_count [ ] [ MACHINE_NUM ] , struct hrd_ctrl_blk * cb , uint16_t debug_ptr ) { uint16_t i , send_ack_count = 0 , remote_clt_id ; struct ibv_wc signal_send_wc ; for ( i = 0 ; i < BCAST_TO_CACHE_BATCH ; i ++ ) { if ( inv_to_send_ops [ i ] . opcode != CACHE_OP_BRC ) { continue ; } if ( credits [ ACK_VC ] [ inv_to_send_ops [ i ] . key . meta . cid ] == 0 ) poll_credits_LIN ( cb -> dgram_recv_cq [ FC_UD_QP_ID ] , credit_wc , credits ) ; if ( credits [ ACK_VC ] [ inv_to_send_ops [ i ] . key . meta . cid ] == 0 ) continue ; credits [ ACK_VC ] [ inv_to_send_ops [ i ] . key . meta . cid ] -- ; remote_clt_id = inv_to_send_ops [ i ] . key . meta . cid * CLIENTS_PER_MACHINE + local_client_id ; coh_message_count [ INV_VC ] [ inv_to_send_ops [ i ] . key . meta . cid ] ++ ; ack_wr [ send_ack_count ] . wr . ud . ah = remote_clt_qp [ remote_clt_id ] [ BROADCAST_UD_QP_ID ] . ah ; ack_wr [ send_ack_count ] . wr . ud . remote_qpn = remote_clt_qp [ remote_clt_id ] [ BROADCAST_UD_QP_ID ] . qpn ; inv_to_send_ops [ i ] . opcode = CACHE_OP_ACK ; inv_to_send_ops [ i ] . key . meta . cid = machine_id ; ack_sgl [ send_ack_count ] . addr = ( uint64_t ) ( uintptr_t ) ( inv_to_send_ops + i ) ; if ( send_ack_count > 0 ) ack_wr [ send_ack_count - 1 ] . next = & ack_wr [ send_ack_count ] ; ack_wr [ send_ack_count ] . send_flags = IBV_SEND_INLINE ; if ( ( * sent_ack_tx ) % ACK_SS_BATCH == 0 ) { ack_wr [ send_ack_count ] . send_flags |= IBV_SEND_SIGNALED ; } if ( ( * sent_ack_tx ) % ACK_SS_BATCH == ACK_SS_BATCH - 1 ) { hrd_poll_cq ( cb -> dgram_send_cq [ BROADCAST_UD_QP_ID ] , 1 , & signal_send_wc ) ; } send_ack_count ++ ; ( * inv_size ) -- ; ( * sent_ack_tx ) ++ ; c_stats [ local_client_id ] . acks_per_client ++ ; if ( ( * sent_ack_tx ) % CREDITS_IN_MESSAGE == 0 ) ( * ack_recv_counter ) ++ ; } post_recvs_and_send ( * ack_recv_counter , send_ack_count , ack_wr , credit_recv_wr , cb -> dgram_qp [ BROADCAST_UD_QP_ID ] , cb -> dgram_qp [ FC_UD_QP_ID ] , MAX_CREDIT_RECVS , BCAST_TO_CACHE_BATCH ) ; if ( * ack_recv_counter > 0 ) * ack_recv_counter = 0 ; }
Pick up the buffer with invalidations that succeeded and respond to them with acks
Pick up the buffer with invalidations that succeeded and respond to them with acks
//      green_printf(" Sending an ack for key %d with version %d \n", //			   inv_to_send_ops[i].key.tag, inv_to_send_ops[i].key.meta.version); // we are emptying that buffer space //		printf("inv %d/%d counter %d,  cid: %d, inv_size %d, credits %d, debug_ptr %d \n", i, BCAST_TO_CACHE_BATCH, //					 coh_message_count[INV_VC][inv_to_send_ops[i].key.meta.cid], inv_to_send_ops[i].key.meta.cid, //					 (*inv_size), credits[ACK_VC][inv_to_send_ops[i].key.meta.cid], debug_ptr); // if (local_client_id == 0) green_printf("Sending ack %llu signaled \n", *sent_ack_tx); // if (local_client_id == 0) green_printf("Polling for ack  %llu \n", *sent_ack_tx); // Selective signaling // Post a receive for a credit every CREDITS_IN_MESSAGE acks // post receives for credits and then send the acks