include/ccKVS/inline_util.h
static inline void updates_and_acks_bookkeeping(uint16_t update_ops_i, struct cache_op* update_ops, struct latency_flags* latency_info,
												struct extended_cache_op* ops, struct timespec* start,
												uint16_t local_client_id, struct mica_resp* resp,
												struct mica_resp* update_resp, uint16_t coh_message_count[][MACHINE_NUM], struct cache_op* ack_bcast_ops,
												uint16_t* ack_push_ptr, uint16_t* ack_size)
{
	uint16_t i;
	for (i = 0; i < update_ops_i; i++) {
		// printf("The %s refers to key %d with cid %d and version %d\n", code_to_str(update_ops[i].opcode),
		//  			update_ops[i].key.tag, update_ops[i].key.meta.cid, update_ops[i].key.meta.version);
		if (update_ops[i].opcode == CACHE_OP_UPD) {
			coh_message_count[UPD_VC][update_ops[i].key.meta.cid]++;
			// WAKE UP
			if (ENABLE_WAKE_UP == 1)  wake_up_stalled(i, update_resp, update_ops, ops, resp);
		}
		else if (update_resp[i].type == CACHE_LAST_ACK_SUCCESS) {
			// wake-up logic
			if (ENABLE_WAKE_UP == 1) wake_up_stalled(i, update_resp, update_ops, ops, resp);
			if (MEASURE_LATENCY == 1 && (latency_info->measured_req_flag == HOT_WRITE_REQ))
				report_latency_of_hot_write(i, update_ops, latency_info, ops, start, local_client_id);
			// this assertion means that the queue is full because the update
			// at the push pointer has not been sent yet
			if (ENABLE_ASSERTIONS == 1) assert(ack_bcast_ops[*ack_push_ptr].opcode != CACHE_OP_ACK);
			ack_bcast_ops[*ack_push_ptr] = update_ops[i];
			ack_bcast_ops[*ack_push_ptr].val_len = (HERD_VALUE_SIZE >> SHIFT_BITS);
			HRD_MOD_ADD(*ack_push_ptr, BCAST_TO_CACHE_BATCH);
			(*ack_size)++;
			if (unlikely(*ack_size > BCAST_TO_CACHE_BATCH)) printf("Error: the ack_size is bigger than allowed %d\n", *ack_size);
		}
		else coh_message_count[ACK_VC][update_ops[i].key.meta.cid]++; // ack was not final, buffer space has emptied
	}
}
// Bookkeeping for received updates and acknowledgemts, optionalL wake-up logic and latency measurement
Bookkeeping for received updates and acknowledgemts, optionalL wake-up logic and latency measurement
Bookkeeping for received updates and acknowledgemts optionalL wake - up logic and latency measurement
static inline void updates_and_acks_bookkeeping(uint16_t update_ops_i, struct cache_op* update_ops, struct latency_flags* latency_info,
												struct extended_cache_op* ops, struct timespec* start,
												uint16_t local_client_id, struct mica_resp* resp,
												struct mica_resp* update_resp, uint16_t coh_message_count[][MACHINE_NUM], struct cache_op* ack_bcast_ops,
												uint16_t* ack_push_ptr, uint16_t* ack_size)
{
	uint16_t i;
	for (i = 0; i < update_ops_i; i++) {
		if (update_ops[i].opcode == CACHE_OP_UPD) {
			coh_message_count[UPD_VC][update_ops[i].key.meta.cid]++;
			if (ENABLE_WAKE_UP == 1)  wake_up_stalled(i, update_resp, update_ops, ops, resp);
		}
		else if (update_resp[i].type == CACHE_LAST_ACK_SUCCESS) {
			if (ENABLE_WAKE_UP == 1) wake_up_stalled(i, update_resp, update_ops, ops, resp);
			if (MEASURE_LATENCY == 1 && (latency_info->measured_req_flag == HOT_WRITE_REQ))
				report_latency_of_hot_write(i, update_ops, latency_info, ops, start, local_client_id);
			if (ENABLE_ASSERTIONS == 1) assert(ack_bcast_ops[*ack_push_ptr].opcode != CACHE_OP_ACK);
			ack_bcast_ops[*ack_push_ptr] = update_ops[i];
			ack_bcast_ops[*ack_push_ptr].val_len = (HERD_VALUE_SIZE >> SHIFT_BITS);
			HRD_MOD_ADD(*ack_push_ptr, BCAST_TO_CACHE_BATCH);
			(*ack_size)++;
			if (unlikely(*ack_size > BCAST_TO_CACHE_BATCH)) printf("Error: the ack_size is bigger than allowed %d\n", *ack_size);
		}
		else coh_message_count[ACK_VC][update_ops[i].key.meta.cid]++; 
	}
}
static inline void updates_and_acks_bookkeeping ( uint16_t update_ops_i , struct cache_op * update_ops , struct latency_flags * latency_info , struct extended_cache_op * ops , struct timespec * start , uint16_t local_client_id , struct mica_resp * resp , struct mica_resp * update_resp , uint16_t coh_message_count [ ] [ MACHINE_NUM ] , struct cache_op * ack_bcast_ops , uint16_t * ack_push_ptr , uint16_t * ack_size ) { uint16_t i ; for ( i = 0 ; i < update_ops_i ; i ++ ) { if ( update_ops [ i ] . opcode == CACHE_OP_UPD ) { coh_message_count [ UPD_VC ] [ update_ops [ i ] . key . meta . cid ] ++ ; if ( ENABLE_WAKE_UP == 1 ) wake_up_stalled ( i , update_resp , update_ops , ops , resp ) ; } else if ( update_resp [ i ] . type == CACHE_LAST_ACK_SUCCESS ) { if ( ENABLE_WAKE_UP == 1 ) wake_up_stalled ( i , update_resp , update_ops , ops , resp ) ; if ( MEASURE_LATENCY == 1 && ( latency_info -> measured_req_flag == HOT_WRITE_REQ ) ) report_latency_of_hot_write ( i , update_ops , latency_info , ops , start , local_client_id ) ; if ( ENABLE_ASSERTIONS == 1 ) assert ( ack_bcast_ops [ * ack_push_ptr ] . opcode != CACHE_OP_ACK ) ; ack_bcast_ops [ * ack_push_ptr ] = update_ops [ i ] ; ack_bcast_ops [ * ack_push_ptr ] . val_len = ( HERD_VALUE_SIZE >> SHIFT_BITS ) ; HRD_MOD_ADD ( * ack_push_ptr , BCAST_TO_CACHE_BATCH ) ; ( * ack_size ) ++ ; if ( unlikely ( * ack_size > BCAST_TO_CACHE_BATCH ) ) printf ( " \n " , * ack_size ) ; } else coh_message_count [ ACK_VC ] [ update_ops [ i ] . key . meta . cid ] ++ ; } }
Bookkeeping for received updates and acknowledgemts, optionalL wake-up logic and latency measurement
Bookkeeping for received updates and acknowledgemts optionalL wake - up logic and latency measurement
// printf("The %s refers to key %d with cid %d and version %d\n", code_to_str(update_ops[i].opcode), //  			update_ops[i].key.tag, update_ops[i].key.meta.cid, update_ops[i].key.meta.version); // WAKE UP // wake-up logic // this assertion means that the queue is full because the update // at the push pointer has not been sent yet // ack was not final, buffer space has emptied