include/ccKVS/inline_util.h
static inline void issue_locals(uint16_t wn, uint16_t* ws, uint16_t local_client_id, uint32_t size_of_op, uint16_t worker_id,
								uint8_t* per_worker_outstanding, long long* remote_for_each_worker, 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 op_i, uint16_t* next_op_i,
								uint16_t local_worker_id, struct latency_flags* latency_info, struct timespec* start,
								struct local_latency* local_measure)
{
	if (ENABLE_LOCAL_WORKERS) {
		wn = local_worker_id;
		int rm_id = worker_id / WORKERS_PER_MACHINE;
		worker_id = rm_id * WORKERS_PER_MACHINE + wn;
	}
	else if (ENABLE_WORKERS_CRCW && ENABLE_STATIC_LOCAL_ALLOCATION) {
		wn = local_client_id % WORKERS_PER_MACHINE;
		int rm_id = worker_id / WORKERS_PER_MACHINE;
		worker_id = rm_id * WORKERS_PER_MACHINE + wn;
	}
	if (DISABLE_LOCALS == 1 && ENABLE_ASSERTIONS == 1) assert (false);
	int offset = OFFSET(wn, local_client_id, ws[wn]);
	uint8_t loc_region = (offset % LOCAL_WINDOW) / LOCAL_REGION_SIZE;

	if (local_recv_flag[wn][local_client_id][loc_region] == 0) {
		if ((MEASURE_LATENCY == 1) && ((latency_info->measured_req_flag) == LOCAL_REQ) &&
			(op_i == (latency_info->last_measured_op_i))) {
			local_measure->measured_local_region = loc_region;
			//			yellow_printf("Issuing a local req: op_i: %d, local_region %d \n",
			//										latency_info->last_measured_op_i, local_measure->measured_local_region);
		}
		memcpy((struct mica_op*) local_req_region + offset, ops + op_i, size_of_op);
		//		printf("op opcode %d local opcode %d, size_of op %d \n", ops[op_i].opcode, local_req_region[offset].opcode, size_of_op);
		if (ENABLE_ASSERTIONS) assert(local_req_region[offset].opcode == MICA_OP_GET ||
									  local_req_region[offset].opcode == MICA_OP_PUT);
		if (ws[wn] % LOCAL_REGION_SIZE == LOCAL_REGION_SIZE - 1) {
			local_recv_flag[wn][local_client_id][loc_region] = 1;
			if ((MEASURE_LATENCY == 1) && ((latency_info->measured_req_flag) == LOCAL_REQ) &&
				((local_measure->measured_local_region) == (loc_region))) {
				local_measure->local_latency_start_polling = 1;
				local_measure->flag_to_poll = (char*)(&local_recv_flag[wn][local_client_id][loc_region]);
				//				yellow_printf("Enable polling for local req on region %d, flag ptr %llu \n ",
				//											local_measure->measured_local_region, local_measure->flag_to_poll);
			}
			// yellow_printf("Lock: client  %d,  region %d for worker %d\n", local_client_id, loc_region, wn);
		}
		// yellow_printf("Client %d I issue a local req to worker %d with opcode %d at offset %d, region %d\n",
		// clt_gid, wn, ops[op_i].opcode, offset, loc_region);

		if (ws[wn] == 0)  per_worker_outstanding[worker_id] = 1;
		else per_worker_outstanding[worker_id]++;
		remote_for_each_worker[worker_id]++;
		c_stats[local_client_id].locals_per_client++;
		HRD_MOD_ADD(ws[wn], LOCAL_WINDOW);
		resp[op_i].type = EMPTY;
	}
	else {
		copy_op_to_next_op(ops, next_ops, resp, next_resp, key_homes, next_key_homes, op_i, next_op_i, latency_info, NULL);
	}
}
// Client logic to issue the local requests
Client logic to issue the local requests
Client logic to issue the local requests
static inline void issue_locals(uint16_t wn, uint16_t* ws, uint16_t local_client_id, uint32_t size_of_op, uint16_t worker_id,
								uint8_t* per_worker_outstanding, long long* remote_for_each_worker, 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 op_i, uint16_t* next_op_i,
								uint16_t local_worker_id, struct latency_flags* latency_info, struct timespec* start,
								struct local_latency* local_measure)
{
	if (ENABLE_LOCAL_WORKERS) {
		wn = local_worker_id;
		int rm_id = worker_id / WORKERS_PER_MACHINE;
		worker_id = rm_id * WORKERS_PER_MACHINE + wn;
	}
	else if (ENABLE_WORKERS_CRCW && ENABLE_STATIC_LOCAL_ALLOCATION) {
		wn = local_client_id % WORKERS_PER_MACHINE;
		int rm_id = worker_id / WORKERS_PER_MACHINE;
		worker_id = rm_id * WORKERS_PER_MACHINE + wn;
	}
	if (DISABLE_LOCALS == 1 && ENABLE_ASSERTIONS == 1) assert (false);
	int offset = OFFSET(wn, local_client_id, ws[wn]);
	uint8_t loc_region = (offset % LOCAL_WINDOW) / LOCAL_REGION_SIZE;
	if (local_recv_flag[wn][local_client_id][loc_region] == 0) {
		if ((MEASURE_LATENCY == 1) && ((latency_info->measured_req_flag) == LOCAL_REQ) &&
			(op_i == (latency_info->last_measured_op_i))) {
			local_measure->measured_local_region = loc_region;
		}
		memcpy((struct mica_op*) local_req_region + offset, ops + op_i, size_of_op);
		if (ENABLE_ASSERTIONS) assert(local_req_region[offset].opcode == MICA_OP_GET ||
									  local_req_region[offset].opcode == MICA_OP_PUT);
		if (ws[wn] % LOCAL_REGION_SIZE == LOCAL_REGION_SIZE - 1) {
			local_recv_flag[wn][local_client_id][loc_region] = 1;
			if ((MEASURE_LATENCY == 1) && ((latency_info->measured_req_flag) == LOCAL_REQ) &&
				((local_measure->measured_local_region) == (loc_region))) {
				local_measure->local_latency_start_polling = 1;
				local_measure->flag_to_poll = (char*)(&local_recv_flag[wn][local_client_id][loc_region]);
			}
		}
		if (ws[wn] == 0)  per_worker_outstanding[worker_id] = 1;
		else per_worker_outstanding[worker_id]++;
		remote_for_each_worker[worker_id]++;
		c_stats[local_client_id].locals_per_client++;
		HRD_MOD_ADD(ws[wn], LOCAL_WINDOW);
		resp[op_i].type = EMPTY;
	}
	else {
		copy_op_to_next_op(ops, next_ops, resp, next_resp, key_homes, next_key_homes, op_i, next_op_i, latency_info, NULL);
	}
}
static inline void issue_locals ( uint16_t wn , uint16_t * ws , uint16_t local_client_id , uint32_t size_of_op , uint16_t worker_id , uint8_t * per_worker_outstanding , long long * remote_for_each_worker , 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 op_i , uint16_t * next_op_i , uint16_t local_worker_id , struct latency_flags * latency_info , struct timespec * start , struct local_latency * local_measure ) { if ( ENABLE_LOCAL_WORKERS ) { wn = local_worker_id ; int rm_id = worker_id / WORKERS_PER_MACHINE ; worker_id = rm_id * WORKERS_PER_MACHINE + wn ; } else if ( ENABLE_WORKERS_CRCW && ENABLE_STATIC_LOCAL_ALLOCATION ) { wn = local_client_id % WORKERS_PER_MACHINE ; int rm_id = worker_id / WORKERS_PER_MACHINE ; worker_id = rm_id * WORKERS_PER_MACHINE + wn ; } if ( DISABLE_LOCALS == 1 && ENABLE_ASSERTIONS == 1 ) assert ( false ) ; int offset = OFFSET ( wn , local_client_id , ws [ wn ] ) ; uint8_t loc_region = ( offset % LOCAL_WINDOW ) / LOCAL_REGION_SIZE ; if ( local_recv_flag [ wn ] [ local_client_id ] [ loc_region ] == 0 ) { if ( ( MEASURE_LATENCY == 1 ) && ( ( latency_info -> measured_req_flag ) == LOCAL_REQ ) && ( op_i == ( latency_info -> last_measured_op_i ) ) ) { local_measure -> measured_local_region = loc_region ; } memcpy ( ( struct mica_op * ) local_req_region + offset , ops + op_i , size_of_op ) ; if ( ENABLE_ASSERTIONS ) assert ( local_req_region [ offset ] . opcode == MICA_OP_GET || local_req_region [ offset ] . opcode == MICA_OP_PUT ) ; if ( ws [ wn ] % LOCAL_REGION_SIZE == LOCAL_REGION_SIZE - 1 ) { local_recv_flag [ wn ] [ local_client_id ] [ loc_region ] = 1 ; if ( ( MEASURE_LATENCY == 1 ) && ( ( latency_info -> measured_req_flag ) == LOCAL_REQ ) && ( ( local_measure -> measured_local_region ) == ( loc_region ) ) ) { local_measure -> local_latency_start_polling = 1 ; local_measure -> flag_to_poll = ( char * ) ( & local_recv_flag [ wn ] [ local_client_id ] [ loc_region ] ) ; } } if ( ws [ wn ] == 0 ) per_worker_outstanding [ worker_id ] = 1 ; else per_worker_outstanding [ worker_id ] ++ ; remote_for_each_worker [ worker_id ] ++ ; c_stats [ local_client_id ] . locals_per_client ++ ; HRD_MOD_ADD ( ws [ wn ] , LOCAL_WINDOW ) ; resp [ op_i ] . type = EMPTY ; } else { copy_op_to_next_op ( ops , next_ops , resp , next_resp , key_homes , next_key_homes , op_i , next_op_i , latency_info , NULL ) ; } }
Client logic to issue the local requests
Client logic to issue the local requests
//			yellow_printf("Issuing a local req: op_i: %d, local_region %d \n", //										latency_info->last_measured_op_i, local_measure->measured_local_region); //		printf("op opcode %d local opcode %d, size_of op %d \n", ops[op_i].opcode, local_req_region[offset].opcode, size_of_op); //				yellow_printf("Enable polling for local req on region %d, flag ptr %llu \n ", //											local_measure->measured_local_region, local_measure->flag_to_poll); // yellow_printf("Lock: client  %d,  region %d for worker %d\n", local_client_id, loc_region, wn); // yellow_printf("Client %d I issue a local req to worker %d with opcode %d at offset %d, region %d\n", // clt_gid, wn, ops[op_i].opcode, offset, loc_region);