include/ccKVS/inline_util.h
static inline uint16_t handle_cold_requests(struct extended_cache_op* ops, struct extended_cache_op* next_ops, struct mica_resp* resp,
											struct mica_resp* next_resp, struct key_home* key_homes, struct key_home* next_key_homes,
											uint16_t* rem_req_i, uint16_t* next_op_i, struct hrd_ctrl_blk* cb, struct ibv_send_wr* rem_send_wr,
											struct ibv_sge* rem_send_sgl, struct ibv_wc* wc, long long remote_tot_tx, uint16_t worker_qp_i,
											uint8_t* per_worker_outstanding, uint32_t* outstanding_rem_reqs, long long* remote_for_each_worker,
											uint16_t* ws, int clt_gid, uint16_t local_client_id, uint16_t* stalled_ops_i, uint16_t local_worker_id,
											int protocol, struct latency_flags* latency_info, struct timespec* start, struct local_latency* local_measure,
											uint16_t* hottest_keys_pointers)
{
	uint16_t op_i = 0, i, j, wr_i = 0, rm_id, wn, worker_id;
	uint32_t size_of_op;
	struct coalesce_inf coalesce_struct[MACHINE_NUM] = {0};
	while ((*rem_req_i) < WINDOW_SIZE && op_i < CACHE_BATCH_SIZE) {
		if (DISABLE_CACHE != 1) {
			if (protocol == LINEARIZABILITY)
				cache_hit_bookkeeping_LIN(ops, next_ops, resp, next_resp, key_homes, next_key_homes,
										  &op_i, next_op_i, stalled_ops_i, local_client_id, latency_info,
										  hottest_keys_pointers);
			else if (protocol == SEQUENTIAL_CONSISTENCY)
				cache_hit_bookkeeping_SC(ops, next_ops, resp, next_resp, key_homes, next_key_homes,
										 &op_i, next_op_i, local_client_id, latency_info,
										 hottest_keys_pointers);
		}
		//Check to see whether the batch of reqs is over
		if (op_i >= CACHE_BATCH_SIZE) {
			if (wr_i > 0) rem_send_wr[wr_i - 1].next = NULL;
			break;
		}

		//Error checking
		if (ENABLE_ASSERTIONS == 1) {
			if (ops[op_i].opcode != CACHE_OP_GET && ops[op_i].opcode != CACHE_OP_PUT)
				printf("Wrong Opcode %d, pointer %d\n", ops[op_i].opcode, op_i);
			if (ops[op_i].opcode == CACHE_OP_PUT) assert(ops[op_i].val_len == (HERD_VALUE_SIZE >> SHIFT_BITS));
		}

		resp[op_i].type = UNSERVED_CACHE_MISS;
		rm_id = key_homes[op_i].machine;	/* Choose a remote machine */
		if (ENABLE_ASSERTIONS == 1) assert(DISABLE_LOCALS == 0 || rm_id != machine_id);
		if (ENABLE_THREAD_PARTITIONING_C_TO_W == 1) wn = (local_client_id + machine_id) % ACTIVE_WORKERS_PER_MACHINE;//
		else wn = key_homes[op_i].worker % ACTIVE_WORKERS_PER_MACHINE;	/* Choose a worker */
		worker_id = rm_id * WORKERS_PER_MACHINE + wn;
		size_of_op = ops[op_i].opcode == CACHE_OP_PUT ? HERD_PUT_REQ_SIZE : HERD_GET_REQ_SIZE;


		/*------------------------------ LOCAL REQUESTS--------------------------------*/
		if (rm_id == machine_id) {
			issue_locals(wn, ws, local_client_id, size_of_op, worker_id, per_worker_outstanding,
						 remote_for_each_worker, ops, next_ops, resp, next_resp, key_homes, next_key_homes,
						 op_i, next_op_i, local_worker_id, latency_info,
						 start, local_measure);
			op_i++;
			continue;
		}
		// overload part of the key to write the clt_gid
		struct remote_meta* tmp =  (struct remote_meta*) &ops[op_i].key.meta;
		tmp->clt_gid = clt_gid;


		//---------------------------------DO COALESCING-----------------------------------------
		if (ENABLE_COALESCING) {
			if (vector_read(ops, coalesce_struct, &op_i, wr_i, rm_id, size_of_op, rem_send_sgl, resp, local_client_id, per_worker_outstanding[worker_id], rem_send_wr)) {
				(*rem_req_i)++;
				continue;
			}
		}

		per_worker_outstanding[worker_id]++;
		if (per_worker_outstanding[worker_id] > WS_PER_WORKER ) {
			//			printf("outstanding %d for worker %d \n", per_worker_outstanding[worker_id], worker_id);
			enum control_flow_directive cf = handle_lack_of_credits_for_remotes(ops, next_ops, resp,
																				next_resp, key_homes, next_key_homes, &op_i, next_op_i, local_client_id, rem_send_wr,
																				&wn, &worker_id, rm_id, per_worker_outstanding, wr_i, *outstanding_rem_reqs, latency_info);
			if (cf == continue_) continue;
			else if (cf == break_) break;
		}

		remote_for_each_worker[worker_id]++;
		/* Forge the RDMA work request */
		forge_remote_wr(wr_i, op_i, wn, rm_id, local_client_id, size_of_op, cb, rem_send_wr, rem_send_sgl, ops,
						wc, remote_tot_tx, worker_id, worker_qp_i);

		wr_i++;
		(*rem_req_i)++;
		resp[op_i].type = EMPTY;
		op_i++;
	} // end while() for remotes

	// Run through the rest of the ops to remove the cache hits here
	if (protocol == LINEARIZABILITY)
		run_through_rest_of_ops_LIN(ops, next_ops, resp, next_resp, key_homes, next_key_homes,
									&op_i, next_op_i, stalled_ops_i, local_client_id, latency_info,
									hottest_keys_pointers);
	else if (protocol == SEQUENTIAL_CONSISTENCY)
		run_through_rest_of_ops_SC(ops, next_ops, resp, next_resp, key_homes, next_key_homes,
								   &op_i, next_op_i, local_client_id, latency_info,
								   hottest_keys_pointers);
	poll_local_req_for_latency_measurement(latency_info, start, local_measure);

	return wr_i;
}
// Handle all cold requests: issue the locals erquests and create the remote work requests
// move to the next op buffer all requests that can not be serviced this cycle
Handle all cold requests: issue the locals erquests and create the remote work requests
move to the next op buffer all requests that can not be serviced this cycle
Handle all cold requests : issue the locals erquests and create the remote work requests move to the next op buffer all requests that can not be serviced this cycle
static inline uint16_t handle_cold_requests(struct extended_cache_op* ops, struct extended_cache_op* next_ops, struct mica_resp* resp,
											struct mica_resp* next_resp, struct key_home* key_homes, struct key_home* next_key_homes,
											uint16_t* rem_req_i, uint16_t* next_op_i, struct hrd_ctrl_blk* cb, struct ibv_send_wr* rem_send_wr,
											struct ibv_sge* rem_send_sgl, struct ibv_wc* wc, long long remote_tot_tx, uint16_t worker_qp_i,
											uint8_t* per_worker_outstanding, uint32_t* outstanding_rem_reqs, long long* remote_for_each_worker,
											uint16_t* ws, int clt_gid, uint16_t local_client_id, uint16_t* stalled_ops_i, uint16_t local_worker_id,
											int protocol, struct latency_flags* latency_info, struct timespec* start, struct local_latency* local_measure,
											uint16_t* hottest_keys_pointers)
{
	uint16_t op_i = 0, i, j, wr_i = 0, rm_id, wn, worker_id;
	uint32_t size_of_op;
	struct coalesce_inf coalesce_struct[MACHINE_NUM] = {0};
	while ((*rem_req_i) < WINDOW_SIZE && op_i < CACHE_BATCH_SIZE) {
		if (DISABLE_CACHE != 1) {
			if (protocol == LINEARIZABILITY)
				cache_hit_bookkeeping_LIN(ops, next_ops, resp, next_resp, key_homes, next_key_homes,
										  &op_i, next_op_i, stalled_ops_i, local_client_id, latency_info,
										  hottest_keys_pointers);
			else if (protocol == SEQUENTIAL_CONSISTENCY)
				cache_hit_bookkeeping_SC(ops, next_ops, resp, next_resp, key_homes, next_key_homes,
										 &op_i, next_op_i, local_client_id, latency_info,
										 hottest_keys_pointers);
		}
		if (op_i >= CACHE_BATCH_SIZE) {
			if (wr_i > 0) rem_send_wr[wr_i - 1].next = NULL;
			break;
		}
		if (ENABLE_ASSERTIONS == 1) {
			if (ops[op_i].opcode != CACHE_OP_GET && ops[op_i].opcode != CACHE_OP_PUT)
				printf("Wrong Opcode %d, pointer %d\n", ops[op_i].opcode, op_i);
			if (ops[op_i].opcode == CACHE_OP_PUT) assert(ops[op_i].val_len == (HERD_VALUE_SIZE >> SHIFT_BITS));
		}
		resp[op_i].type = UNSERVED_CACHE_MISS;
		rm_id = key_homes[op_i].machine;	
		if (ENABLE_ASSERTIONS == 1) assert(DISABLE_LOCALS == 0 || rm_id != machine_id);
		if (ENABLE_THREAD_PARTITIONING_C_TO_W == 1) wn = (local_client_id + machine_id) % ACTIVE_WORKERS_PER_MACHINE;
		else wn = key_homes[op_i].worker % ACTIVE_WORKERS_PER_MACHINE;	
		worker_id = rm_id * WORKERS_PER_MACHINE + wn;
		size_of_op = ops[op_i].opcode == CACHE_OP_PUT ? HERD_PUT_REQ_SIZE : HERD_GET_REQ_SIZE;
		if (rm_id == machine_id) {
			issue_locals(wn, ws, local_client_id, size_of_op, worker_id, per_worker_outstanding,
						 remote_for_each_worker, ops, next_ops, resp, next_resp, key_homes, next_key_homes,
						 op_i, next_op_i, local_worker_id, latency_info,
						 start, local_measure);
			op_i++;
			continue;
		}
		 overload part of the key to write the clt_gid
		struct remote_meta* tmp =  (struct remote_meta*) &ops[op_i].key.meta;
		tmp->clt_gid = clt_gid;
		---------------------------------DO COALESCING-----------------------------------------
		if (ENABLE_COALESCING) {
			if (vector_read(ops, coalesce_struct, &op_i, wr_i, rm_id, size_of_op, rem_send_sgl, resp, local_client_id, per_worker_outstanding[worker_id], rem_send_wr)) {
				(*rem_req_i)++;
				continue;
			}
		}
		per_worker_outstanding[worker_id]++;
		if (per_worker_outstanding[worker_id] > WS_PER_WORKER ) {
						printf("outstanding %d for worker %d \n", per_worker_outstanding[worker_id], worker_id);
			enum control_flow_directive cf = handle_lack_of_credits_for_remotes(ops, next_ops, resp,
																				next_resp, key_homes, next_key_homes, &op_i, next_op_i, local_client_id, rem_send_wr,
																				&wn, &worker_id, rm_id, per_worker_outstanding, wr_i, *outstanding_rem_reqs, latency_info);
			if (cf == continue_) continue;
			else if (cf == break_) break;
		}
		remote_for_each_worker[worker_id]++;
		forge_remote_wr(wr_i, op_i, wn, rm_id, local_client_id, size_of_op, cb, rem_send_wr, rem_send_sgl, ops,
						wc, remote_tot_tx, worker_id, worker_qp_i);
		wr_i++;
		(*rem_req_i)++;
		resp[op_i].type = EMPTY;
		op_i++;
	}  end while() for remotes
	 Run through the rest of the ops to remove the cache hits here
	if (protocol == LINEARIZABILITY)
		run_through_rest_of_ops_LIN(ops, next_ops, resp, next_resp, key_homes, next_key_homes,
									&op_i, next_op_i, stalled_ops_i, local_client_id, latency_info,
									hottest_keys_pointers);
	else if (protocol == SEQUENTIAL_CONSISTENCY)
		run_through_rest_of_ops_SC(ops, next_ops, resp, next_resp, key_homes, next_key_homes,
								   &op_i, next_op_i, local_client_id, latency_info,
								   hottest_keys_pointers);
	poll_local_req_for_latency_measurement(latency_info, start, local_measure);
	return wr_i;
}
static inline uint16_t handle_cold_requests ( struct extended_cache_op * ops , struct extended_cache_op * next_ops , struct mica_resp * resp , struct mica_resp * next_resp , struct key_home * key_homes , struct key_home * next_key_homes , uint16_t * rem_req_i , uint16_t * next_op_i , struct hrd_ctrl_blk * cb , struct ibv_send_wr * rem_send_wr , struct ibv_sge * rem_send_sgl , struct ibv_wc * wc , long long remote_tot_tx , uint16_t worker_qp_i , uint8_t * per_worker_outstanding , uint32_t * outstanding_rem_reqs , long long * remote_for_each_worker , uint16_t * ws , int clt_gid , uint16_t local_client_id , uint16_t * stalled_ops_i , uint16_t local_worker_id , int protocol , struct latency_flags * latency_info , struct timespec * start , struct local_latency * local_measure , uint16_t * hottest_keys_pointers ) { uint16_t op_i = 0 , i , j , wr_i = 0 , rm_id , wn , worker_id ; uint32_t size_of_op ; struct coalesce_inf coalesce_struct [ MACHINE_NUM ] = { 0 } ; while ( ( * rem_req_i ) < WINDOW_SIZE && op_i < CACHE_BATCH_SIZE ) { if ( DISABLE_CACHE != 1 ) { if ( protocol == LINEARIZABILITY ) cache_hit_bookkeeping_LIN ( ops , next_ops , resp , next_resp , key_homes , next_key_homes , & op_i , next_op_i , stalled_ops_i , local_client_id , latency_info , hottest_keys_pointers ) ; else if ( protocol == SEQUENTIAL_CONSISTENCY ) cache_hit_bookkeeping_SC ( ops , next_ops , resp , next_resp , key_homes , next_key_homes , & op_i , next_op_i , local_client_id , latency_info , hottest_keys_pointers ) ; } if ( op_i >= CACHE_BATCH_SIZE ) { if ( wr_i > 0 ) rem_send_wr [ wr_i - 1 ] . next = NULL ; break ; } if ( ENABLE_ASSERTIONS == 1 ) { if ( ops [ op_i ] . opcode != CACHE_OP_GET && ops [ op_i ] . opcode != CACHE_OP_PUT ) printf ( " \n " , ops [ op_i ] . opcode , op_i ) ; if ( ops [ op_i ] . opcode == CACHE_OP_PUT ) assert ( ops [ op_i ] . val_len == ( HERD_VALUE_SIZE >> SHIFT_BITS ) ) ; } resp [ op_i ] . type = UNSERVED_CACHE_MISS ; rm_id = key_homes [ op_i ] . machine ; if ( ENABLE_ASSERTIONS == 1 ) assert ( DISABLE_LOCALS == 0 || rm_id != machine_id ) ; if ( ENABLE_THREAD_PARTITIONING_C_TO_W == 1 ) wn = ( local_client_id + machine_id ) % ACTIVE_WORKERS_PER_MACHINE ; else wn = key_homes [ op_i ] . worker % ACTIVE_WORKERS_PER_MACHINE ; worker_id = rm_id * WORKERS_PER_MACHINE + wn ; size_of_op = ops [ op_i ] . opcode == CACHE_OP_PUT ? HERD_PUT_REQ_SIZE : HERD_GET_REQ_SIZE ; if ( rm_id == machine_id ) { issue_locals ( wn , ws , local_client_id , size_of_op , worker_id , per_worker_outstanding , remote_for_each_worker , ops , next_ops , resp , next_resp , key_homes , next_key_homes , op_i , next_op_i , local_worker_id , latency_info , start , local_measure ) ; op_i ++ ; continue ; } struct remote_meta * tmp = ( struct remote_meta * ) & ops [ op_i ] . key . meta ; tmp -> clt_gid = clt_gid ; if ( ENABLE_COALESCING ) { if ( vector_read ( ops , coalesce_struct , & op_i , wr_i , rm_id , size_of_op , rem_send_sgl , resp , local_client_id , per_worker_outstanding [ worker_id ] , rem_send_wr ) ) { ( * rem_req_i ) ++ ; continue ; } } per_worker_outstanding [ worker_id ] ++ ; if ( per_worker_outstanding [ worker_id ] > WS_PER_WORKER ) { enum control_flow_directive cf = handle_lack_of_credits_for_remotes ( ops , next_ops , resp , next_resp , key_homes , next_key_homes , & op_i , next_op_i , local_client_id , rem_send_wr , & wn , & worker_id , rm_id , per_worker_outstanding , wr_i , * outstanding_rem_reqs , latency_info ) ; if ( cf == continue_ ) continue ; else if ( cf == break_ ) break ; } remote_for_each_worker [ worker_id ] ++ ; forge_remote_wr ( wr_i , op_i , wn , rm_id , local_client_id , size_of_op , cb , rem_send_wr , rem_send_sgl , ops , wc , remote_tot_tx , worker_id , worker_qp_i ) ; wr_i ++ ; ( * rem_req_i ) ++ ; resp [ op_i ] . type = EMPTY ; op_i ++ ; } if ( protocol == LINEARIZABILITY ) run_through_rest_of_ops_LIN ( ops , next_ops , resp , next_resp , key_homes , next_key_homes , & op_i , next_op_i , stalled_ops_i , local_client_id , latency_info , hottest_keys_pointers ) ; else if ( protocol == SEQUENTIAL_CONSISTENCY ) run_through_rest_of_ops_SC ( ops , next_ops , resp , next_resp , key_homes , next_key_homes , & op_i , next_op_i , local_client_id , latency_info , hottest_keys_pointers ) ; poll_local_req_for_latency_measurement ( latency_info , start , local_measure ) ; return wr_i ; }
Handle all cold requests: issue the locals erquests and create the remote work requests
move to the next op buffer all requests that can not be serviced this cycle
Handle all cold requests : issue the locals erquests and create the remote work requests move to the next op buffer all requests that can not be serviced this cycle
//Check to see whether the batch of reqs is over //Error checking /* Choose a remote machine */ // /* Choose a worker */ /*------------------------------ LOCAL REQUESTS--------------------------------*/ // overload part of the key to write the clt_gid //---------------------------------DO COALESCING----------------------------------------- //			printf("outstanding %d for worker %d \n", per_worker_outstanding[worker_id], worker_id); /* Forge the RDMA work request */ // end while() for remotes // Run through the rest of the ops to remove the cache hits here