include/ccKVS/inline_util.h
static inline uint16_t forge_credits_LIN(uint16_t coh_message_count[][MACHINE_NUM], uint16_t *acks_seen, uint16_t *invs_seen,
										 uint16_t *upds_seen, uint16_t local_client_id, struct ibv_send_wr *credit_wr,
										 long long *credit_tx,
										 struct hrd_ctrl_blk *cb, struct ibv_cq *coh_recv_cq, struct ibv_wc *coh_wc)
{
	// Send out credits for the polled requests
	uint16_t total_fc_message_count = 0;
	uint16_t credit_wr_i = 0, rm_id, i, j;
	struct ibv_wc signal_send_wc;
	for (rm_id = 0; rm_id < MACHINE_NUM; rm_id++) {
		uint16_t fc_message_count = coh_message_count[ACK_VC][rm_id] + coh_message_count[INV_VC][rm_id] + coh_message_count[UPD_VC][rm_id];
		if (fc_message_count > 0) {
			total_fc_message_count += fc_message_count;
			for (i = 0; i < fc_message_count; i++) {
				// if (i < coh_message_count[ACK_VC][rm_id]) printf("I received an ack from machine %d\n", rm_id);
				// else if (i < coh_message_count[INV_VC][rm_id] + coh_message_count[ACK_VC][rm_id]) printf("I received an inv from machine %d\n", rm_id);
				// else printf("I received a upd from machine %d\n", rm_id);
				//rm_id = wc[i].imm_data;

				if (i < coh_message_count[ACK_VC][rm_id]) acks_seen[rm_id]++;
				else if (i < coh_message_count[INV_VC][rm_id] + coh_message_count[ACK_VC][rm_id]) invs_seen[rm_id]++;
				else upds_seen[rm_id]++;
				// If we have seen enough broadcasts from a given machine send credits
				if (acks_seen[rm_id] == CREDITS_IN_MESSAGE || invs_seen[rm_id] == CREDITS_IN_MESSAGE || upds_seen[rm_id] == CREDITS_IN_MESSAGE) {
					uint16_t clt_i = rm_id * CLIENTS_PER_MACHINE + local_client_id;
					credit_wr[credit_wr_i].wr.ud.ah = remote_clt_qp[clt_i][FC_UD_QP_ID].ah;
					credit_wr[credit_wr_i].wr.ud.remote_qpn = remote_clt_qp[clt_i][FC_UD_QP_ID].qpn;
					// CONVENTION to denote where the credit refers to, acks have the machine _ids from 0 to MACHINE_NUM - 1
					if (i < coh_message_count[ACK_VC][rm_id] ) { //ack
						credit_wr[credit_wr_i].imm_data = machine_id;
						acks_seen[rm_id] = 0;
						//                            cyan_printf("sending an ack credit to %d\n", rm_id);
					}
					else if (i < coh_message_count[INV_VC][rm_id] + coh_message_count[ACK_VC][rm_id]) { //inv
						credit_wr[credit_wr_i].imm_data = machine_id + MACHINE_NUM;
						invs_seen[rm_id] = 0;
						//                            cyan_printf("sending an inv credit to %d\n", rm_id);
					}
					else { //upd
						credit_wr[credit_wr_i].imm_data = machine_id + (2 * MACHINE_NUM);
						upds_seen[rm_id] = 0;
						//                            cyan_printf("sending an upd credit to %d\n", rm_id);
					}

					if (credit_wr_i > 0) credit_wr[credit_wr_i - 1].next = &credit_wr[credit_wr_i];
					if (((*credit_tx) % CREDIT_SS_BATCH) == 0) {
						credit_wr[credit_wr_i].send_flags |= IBV_SEND_SIGNALED;
					}
					else credit_wr[credit_wr_i].send_flags = IBV_SEND_INLINE;
					if(((*credit_tx) % CREDIT_SS_BATCH) == CREDIT_SS_BATCH - 1) {
						hrd_poll_cq(cb->dgram_send_cq[FC_UD_QP_ID], 1, &signal_send_wc);
					}
					credit_wr_i++;
					(*credit_tx)++;
					if (ENABLE_ASSERTIONS == 1) assert(credit_wr_i <= CREDIT_SS_BATCH_);
				}
			}
		}
	}
	hrd_poll_cq(coh_recv_cq, total_fc_message_count, coh_wc);
	return credit_wr_i;
}
/* Check the buffer space that has been emptied by the previous steps and forge the appropriate credits
   Buffer space empties on: propagating updates, non-final acks, failed invs, sending acks, sending updates */
Check the buffer space that has been emptied by the previous steps and forge the appropriate credits
Buffer space empties on: propagating updates, non-final acks, failed invs, sending acks, sending updates
Check the buffer space that has been emptied by the previous steps and forge the appropriate credits Buffer space empties on : propagating updates non - final acks failed invs sending acks sending updates
static inline uint16_t forge_credits_LIN(uint16_t coh_message_count[][MACHINE_NUM], uint16_t *acks_seen, uint16_t *invs_seen,
										 uint16_t *upds_seen, uint16_t local_client_id, struct ibv_send_wr *credit_wr,
										 long long *credit_tx,
										 struct hrd_ctrl_blk *cb, struct ibv_cq *coh_recv_cq, struct ibv_wc *coh_wc)
{
	uint16_t total_fc_message_count = 0;
	uint16_t credit_wr_i = 0, rm_id, i, j;
	struct ibv_wc signal_send_wc;
	for (rm_id = 0; rm_id < MACHINE_NUM; rm_id++) {
		uint16_t fc_message_count = coh_message_count[ACK_VC][rm_id] + coh_message_count[INV_VC][rm_id] + coh_message_count[UPD_VC][rm_id];
		if (fc_message_count > 0) {
			total_fc_message_count += fc_message_count;
			for (i = 0; i < fc_message_count; i++) {
				if (i < coh_message_count[ACK_VC][rm_id]) acks_seen[rm_id]++;
				else if (i < coh_message_count[INV_VC][rm_id] + coh_message_count[ACK_VC][rm_id]) invs_seen[rm_id]++;
				else upds_seen[rm_id]++;
				if (acks_seen[rm_id] == CREDITS_IN_MESSAGE || invs_seen[rm_id] == CREDITS_IN_MESSAGE || upds_seen[rm_id] == CREDITS_IN_MESSAGE) {
					uint16_t clt_i = rm_id * CLIENTS_PER_MACHINE + local_client_id;
					credit_wr[credit_wr_i].wr.ud.ah = remote_clt_qp[clt_i][FC_UD_QP_ID].ah;
					credit_wr[credit_wr_i].wr.ud.remote_qpn = remote_clt_qp[clt_i][FC_UD_QP_ID].qpn;
					if (i < coh_message_count[ACK_VC][rm_id] ) { 
						credit_wr[credit_wr_i].imm_data = machine_id;
						acks_seen[rm_id] = 0;
					}
					else if (i < coh_message_count[INV_VC][rm_id] + coh_message_count[ACK_VC][rm_id]) { 
						credit_wr[credit_wr_i].imm_data = machine_id + MACHINE_NUM;
						invs_seen[rm_id] = 0;
					}
					else { 
						credit_wr[credit_wr_i].imm_data = machine_id + (2 * MACHINE_NUM);
						upds_seen[rm_id] = 0;
					}
					if (credit_wr_i > 0) credit_wr[credit_wr_i - 1].next = &credit_wr[credit_wr_i];
					if (((*credit_tx) % CREDIT_SS_BATCH) == 0) {
						credit_wr[credit_wr_i].send_flags |= IBV_SEND_SIGNALED;
					}
					else credit_wr[credit_wr_i].send_flags = IBV_SEND_INLINE;
					if(((*credit_tx) % CREDIT_SS_BATCH) == CREDIT_SS_BATCH - 1) {
						hrd_poll_cq(cb->dgram_send_cq[FC_UD_QP_ID], 1, &signal_send_wc);
					}
					credit_wr_i++;
					(*credit_tx)++;
					if (ENABLE_ASSERTIONS == 1) assert(credit_wr_i <= CREDIT_SS_BATCH_);
				}
			}
		}
	}
	hrd_poll_cq(coh_recv_cq, total_fc_message_count, coh_wc);
	return credit_wr_i;
}
static inline uint16_t forge_credits_LIN ( uint16_t coh_message_count [ ] [ MACHINE_NUM ] , uint16_t * acks_seen , uint16_t * invs_seen , uint16_t * upds_seen , uint16_t local_client_id , struct ibv_send_wr * credit_wr , long long * credit_tx , struct hrd_ctrl_blk * cb , struct ibv_cq * coh_recv_cq , struct ibv_wc * coh_wc ) { uint16_t total_fc_message_count = 0 ; uint16_t credit_wr_i = 0 , rm_id , i , j ; struct ibv_wc signal_send_wc ; for ( rm_id = 0 ; rm_id < MACHINE_NUM ; rm_id ++ ) { uint16_t fc_message_count = coh_message_count [ ACK_VC ] [ rm_id ] + coh_message_count [ INV_VC ] [ rm_id ] + coh_message_count [ UPD_VC ] [ rm_id ] ; if ( fc_message_count > 0 ) { total_fc_message_count += fc_message_count ; for ( i = 0 ; i < fc_message_count ; i ++ ) { if ( i < coh_message_count [ ACK_VC ] [ rm_id ] ) acks_seen [ rm_id ] ++ ; else if ( i < coh_message_count [ INV_VC ] [ rm_id ] + coh_message_count [ ACK_VC ] [ rm_id ] ) invs_seen [ rm_id ] ++ ; else upds_seen [ rm_id ] ++ ; if ( acks_seen [ rm_id ] == CREDITS_IN_MESSAGE || invs_seen [ rm_id ] == CREDITS_IN_MESSAGE || upds_seen [ rm_id ] == CREDITS_IN_MESSAGE ) { uint16_t clt_i = rm_id * CLIENTS_PER_MACHINE + local_client_id ; credit_wr [ credit_wr_i ] . wr . ud . ah = remote_clt_qp [ clt_i ] [ FC_UD_QP_ID ] . ah ; credit_wr [ credit_wr_i ] . wr . ud . remote_qpn = remote_clt_qp [ clt_i ] [ FC_UD_QP_ID ] . qpn ; if ( i < coh_message_count [ ACK_VC ] [ rm_id ] ) { credit_wr [ credit_wr_i ] . imm_data = machine_id ; acks_seen [ rm_id ] = 0 ; } else if ( i < coh_message_count [ INV_VC ] [ rm_id ] + coh_message_count [ ACK_VC ] [ rm_id ] ) { credit_wr [ credit_wr_i ] . imm_data = machine_id + MACHINE_NUM ; invs_seen [ rm_id ] = 0 ; } else { credit_wr [ credit_wr_i ] . imm_data = machine_id + ( 2 * MACHINE_NUM ) ; upds_seen [ rm_id ] = 0 ; } if ( credit_wr_i > 0 ) credit_wr [ credit_wr_i - 1 ] . next = & credit_wr [ credit_wr_i ] ; if ( ( ( * credit_tx ) % CREDIT_SS_BATCH ) == 0 ) { credit_wr [ credit_wr_i ] . send_flags |= IBV_SEND_SIGNALED ; } else credit_wr [ credit_wr_i ] . send_flags = IBV_SEND_INLINE ; if ( ( ( * credit_tx ) % CREDIT_SS_BATCH ) == CREDIT_SS_BATCH - 1 ) { hrd_poll_cq ( cb -> dgram_send_cq [ FC_UD_QP_ID ] , 1 , & signal_send_wc ) ; } credit_wr_i ++ ; ( * credit_tx ) ++ ; if ( ENABLE_ASSERTIONS == 1 ) assert ( credit_wr_i <= CREDIT_SS_BATCH_ ) ; } } } } hrd_poll_cq ( coh_recv_cq , total_fc_message_count , coh_wc ) ; return credit_wr_i ; }
Check the buffer space that has been emptied by the previous steps and forge the appropriate credits
Buffer space empties on: propagating updates, non-final acks, failed invs, sending acks, sending updates
Check the buffer space that has been emptied by the previous steps and forge the appropriate credits Buffer space empties on : propagating updates non - final acks failed invs sending acks sending updates
// Send out credits for the polled requests // if (i < coh_message_count[ACK_VC][rm_id]) printf("I received an ack from machine %d\n", rm_id); // else if (i < coh_message_count[INV_VC][rm_id] + coh_message_count[ACK_VC][rm_id]) printf("I received an inv from machine %d\n", rm_id); // else printf("I received a upd from machine %d\n", rm_id); //rm_id = wc[i].imm_data; // If we have seen enough broadcasts from a given machine send credits // CONVENTION to denote where the credit refers to, acks have the machine _ids from 0 to MACHINE_NUM - 1 //ack //                            cyan_printf("sending an ack credit to %d\n", rm_id); //inv //                            cyan_printf("sending an inv credit to %d\n", rm_id); //upd //                            cyan_printf("sending an upd credit to %d\n", rm_id);