src/ccKVS/cache.c
int batch_from_trace_to_cache(int trace_iter, int thread_id, struct trace_command *trace,
							  struct extended_cache_op *ops, struct mica_resp *resp, struct key_home* kh, int isSC, uint16_t next_op_i,
							  struct latency_flags* latency_info, struct timespec* start, uint16_t* hottest_keys_pointers) {
	int i = next_op_i, j = 0;
	uint8_t is_update = 0;
	uint64_t seed = 0xdeadbeef;
	uint32_t empty_reqs = 0;

	if (ENABLE_ASSERTIONS == 1) {
		for (j = 0; j < next_op_i; j++){
			if (resp[j].type == EMPTY) {
				printf("Client %d, j = %d\n",thread_id, j );
				assert(false);}
			assert(!(resp[j].type == CACHE_GET_SUCCESS || ops[j].opcode == CACHE_OP_INV ||
					 resp[j].type == CACHE_PUT_FAIL));
		}
	}
	if ((ENABLE_HOT_KEY_TRACKING == 1) && (isSC == 0)) memset(hottest_keys_pointers, 0, sizeof(uint16_t) * HOTTEST_KEYS_TO_TRACK);
//  green_printf("NEW BUFFER \n");
	while (i < CACHE_BATCH_SIZE && trace[trace_iter].opcode != NOP) {

		if ((ENABLE_HOT_KEY_TRACKING == 1) && (DISABLE_CACHE == 0)) {
			if (trace[trace_iter].key_id < HOTTEST_KEYS_TO_TRACK) {
				if (IS_READ(trace[trace_iter].opcode) == 1) {
					if (hottest_keys_pointers[trace[trace_iter].key_id] == 0) {
						hottest_keys_pointers[trace[trace_iter].key_id] = i + 1; //tag the op_i+1 such that '0' means unused
						if (isSC == 1) ops[i].value[0] = (uint8_t) trace[trace_iter].key_id; //the op tracks the key it stores
//          yellow_printf("Start tracking key %d , in op  %d\n", trace[trace_iter].key_id, i);
					} else {
						uint16_t op_i = hottest_keys_pointers[trace[trace_iter].key_id] - 1;
//            printf("opcode %d, op_i %d key %d key_pointer_to_op %d \n",
//                   ops[op_i].opcode, op_i, trace[trace_iter].key_id, hottest_keys_pointers[trace[trace_iter].key_id]);
						if (ENABLE_ASSERTIONS) {
							assert((hottest_keys_pointers[trace[trace_iter].key_id] > 0));
							assert(ops[op_i].opcode == CACHE_OP_GET);
							assert(ops[op_i].val_len < 255);
						}
						ops[op_i].val_len++;
//          yellow_printf("tracked key %d , in op  %d, new val_len %d\n", trace[trace_iter].key_id, op_i, ops[op_i].val_len);
						trace_iter++;
						if (trace[trace_iter].opcode == NOP)
							trace_iter = 0;
						continue;
					}
				}
				else hottest_keys_pointers[trace[trace_iter].key_id] = 0; // on a write
			}
		}
		*(uint128 *) &ops[i] = trace[trace_iter].key_hash;
		if (MEASURE_LATENCY == 1) ops[i].key.meta.state = 0;
		// *(uint128 *) &ops[i] = CityHash128((char *) &(trace[trace_iter].key_id), 4);
		// printf("Cache: Key = %d, tag: %d. \n", trace[trace_iter].key_id, ops[i].key.tag);
		is_update = (IS_WRITE(trace[trace_iter].opcode)) ? (uint8_t) 1 : (uint8_t) 0;
		if (ENABLE_ASSERTIONS == 1) assert(WRITE_RATIO > 0 || is_update == 0);
		ops[i].opcode = is_update ? (uint8_t) CACHE_OP_PUT : (uint8_t) CACHE_OP_GET;
		ops[i].val_len = is_update ? (uint8_t) (HERD_VALUE_SIZE >> SHIFT_BITS) : (uint8_t) 0; // if it is not an update the val_len will not be included in the packet

		if(is_update)
			if (ENABLE_ASSERTIONS == 1) assert(ops[i].val_len > 0);
		ops[i].key.meta.cid = (uint8_t) ((ENABLE_MULTIPLE_SESSIONS == 1) ?
										 (uint8_t)(hrd_fastrand(&seed) % SESSIONS_PER_CLIENT)
										 * CLIENTS_PER_MACHINE + thread_id : thread_id); // only for multiple sessions

		if (MEASURE_LATENCY == 1) start_measurement(start, latency_info, trace[trace_iter].home_machine_id,
													ops, i, thread_id, trace[trace_iter].opcode, isSC, next_op_i);

		kh[i].machine = (uint8_t) trace[trace_iter].home_machine_id;
		kh[i].worker = (uint8_t) trace[trace_iter].home_worker_id;
		resp[i].type = EMPTY;
		if(ops[i].opcode == CACHE_OP_PUT) //put the folowing value
			str_to_binary(ops[i].value, "Armonia is the key to success!  ", HERD_VALUE_SIZE);
		trace_iter++;
		empty_reqs++;
		if (DISABLE_CACHE == 1) resp[i].type = CACHE_MISS;
		i++;
		if (trace[trace_iter].opcode == NOP)
			trace_iter = 0;
	}
	empty_reqs *= 100;
	c_stats[thread_id].empty_reqs_per_trace = (double) empty_reqs / CACHE_BATCH_SIZE;
	c_stats[thread_id].tot_empty_reqs_per_trace += c_stats[thread_id].empty_reqs_per_trace;
	if (DISABLE_CACHE == 0) {
		if(isSC == 1) cache_batch_op_lin_non_stalling_sessions(i, thread_id, &ops, resp);
		else cache_batch_op_sc(i, thread_id, &ops, resp);
	}
	if (MEASURE_LATENCY == 1 && (DISABLE_CACHE == 0)) {
		hot_request_bookkeeping_for_latency_measurements(start, latency_info, ops, i, thread_id, isSC, resp);
	}

	if (ENABLE_ASSERTIONS == 1) {
		for (j = 0; j < CACHE_BATCH_SIZE; j++) { assert(resp[j].type != RETRY); }
	}
	return trace_iter;
}
//refill empty slots of array from the trace
refill empty slots of array from the trace
refill empty slots of array from the trace
int batch_from_trace_to_cache(int trace_iter, int thread_id, struct trace_command *trace,
							  struct extended_cache_op *ops, struct mica_resp *resp, struct key_home* kh, int isSC, uint16_t next_op_i,
							  struct latency_flags* latency_info, struct timespec* start, uint16_t* hottest_keys_pointers) {
	int i = next_op_i, j = 0;
	uint8_t is_update = 0;
	uint64_t seed = 0xdeadbeef;
	uint32_t empty_reqs = 0;
	if (ENABLE_ASSERTIONS == 1) {
		for (j = 0; j < next_op_i; j++){
			if (resp[j].type == EMPTY) {
				printf("Client %d, j = %d\n",thread_id, j );
				assert(false);}
			assert(!(resp[j].type == CACHE_GET_SUCCESS || ops[j].opcode == CACHE_OP_INV ||
					 resp[j].type == CACHE_PUT_FAIL));
		}
	}
	if ((ENABLE_HOT_KEY_TRACKING == 1) && (isSC == 0)) memset(hottest_keys_pointers, 0, sizeof(uint16_t) * HOTTEST_KEYS_TO_TRACK);
	while (i < CACHE_BATCH_SIZE && trace[trace_iter].opcode != NOP) {
		if ((ENABLE_HOT_KEY_TRACKING == 1) && (DISABLE_CACHE == 0)) {
			if (trace[trace_iter].key_id < HOTTEST_KEYS_TO_TRACK) {
				if (IS_READ(trace[trace_iter].opcode) == 1) {
					if (hottest_keys_pointers[trace[trace_iter].key_id] == 0) {
						hottest_keys_pointers[trace[trace_iter].key_id] = i + 1; 
						if (isSC == 1) ops[i].value[0] = (uint8_t) trace[trace_iter].key_id; 
					} else {
						uint16_t op_i = hottest_keys_pointers[trace[trace_iter].key_id] - 1;
						if (ENABLE_ASSERTIONS) {
							assert((hottest_keys_pointers[trace[trace_iter].key_id] > 0));
							assert(ops[op_i].opcode == CACHE_OP_GET);
							assert(ops[op_i].val_len < 255);
						}
						ops[op_i].val_len++;
						trace_iter++;
						if (trace[trace_iter].opcode == NOP)
							trace_iter = 0;
						continue;
					}
				}
				else hottest_keys_pointers[trace[trace_iter].key_id] = 0; 
			}
		}
		*(uint128 *) &ops[i] = trace[trace_iter].key_hash;
		if (MEASURE_LATENCY == 1) ops[i].key.meta.state = 0;
		is_update = (IS_WRITE(trace[trace_iter].opcode)) ? (uint8_t) 1 : (uint8_t) 0;
		if (ENABLE_ASSERTIONS == 1) assert(WRITE_RATIO > 0 || is_update == 0);
		ops[i].opcode = is_update ? (uint8_t) CACHE_OP_PUT : (uint8_t) CACHE_OP_GET;
		ops[i].val_len = is_update ? (uint8_t) (HERD_VALUE_SIZE >> SHIFT_BITS) : (uint8_t) 0; 
		if(is_update)
			if (ENABLE_ASSERTIONS == 1) assert(ops[i].val_len > 0);
		ops[i].key.meta.cid = (uint8_t) ((ENABLE_MULTIPLE_SESSIONS == 1) ?
										 (uint8_t)(hrd_fastrand(&seed) % SESSIONS_PER_CLIENT)
										 * CLIENTS_PER_MACHINE + thread_id : thread_id); 
		if (MEASURE_LATENCY == 1) start_measurement(start, latency_info, trace[trace_iter].home_machine_id,
													ops, i, thread_id, trace[trace_iter].opcode, isSC, next_op_i);
		kh[i].machine = (uint8_t) trace[trace_iter].home_machine_id;
		kh[i].worker = (uint8_t) trace[trace_iter].home_worker_id;
		resp[i].type = EMPTY;
		if(ops[i].opcode == CACHE_OP_PUT) 
			str_to_binary(ops[i].value, "Armonia is the key to success!  ", HERD_VALUE_SIZE);
		trace_iter++;
		empty_reqs++;
		if (DISABLE_CACHE == 1) resp[i].type = CACHE_MISS;
		i++;
		if (trace[trace_iter].opcode == NOP)
			trace_iter = 0;
	}
	empty_reqs *= 100;
	c_stats[thread_id].empty_reqs_per_trace = (double) empty_reqs / CACHE_BATCH_SIZE;
	c_stats[thread_id].tot_empty_reqs_per_trace += c_stats[thread_id].empty_reqs_per_trace;
	if (DISABLE_CACHE == 0) {
		if(isSC == 1) cache_batch_op_lin_non_stalling_sessions(i, thread_id, &ops, resp);
		else cache_batch_op_sc(i, thread_id, &ops, resp);
	}
	if (MEASURE_LATENCY == 1 && (DISABLE_CACHE == 0)) {
		hot_request_bookkeeping_for_latency_measurements(start, latency_info, ops, i, thread_id, isSC, resp);
	}
	if (ENABLE_ASSERTIONS == 1) {
		for (j = 0; j < CACHE_BATCH_SIZE; j++) { assert(resp[j].type != RETRY); }
	}
	return trace_iter;
}
int batch_from_trace_to_cache ( int trace_iter , int thread_id , struct trace_command * trace , struct extended_cache_op * ops , struct mica_resp * resp , struct key_home * kh , int isSC , uint16_t next_op_i , struct latency_flags * latency_info , struct timespec * start , uint16_t * hottest_keys_pointers ) { int i = next_op_i , j = 0 ; uint8_t is_update = 0 ; uint64_t seed = 0xdeadbeef ; uint32_t empty_reqs = 0 ; if ( ENABLE_ASSERTIONS == 1 ) { for ( j = 0 ; j < next_op_i ; j ++ ) { if ( resp [ j ] . type == EMPTY ) { printf ( " \n " , thread_id , j ) ; assert ( false ) ; } assert ( ! ( resp [ j ] . type == CACHE_GET_SUCCESS || ops [ j ] . opcode == CACHE_OP_INV || resp [ j ] . type == CACHE_PUT_FAIL ) ) ; } } if ( ( ENABLE_HOT_KEY_TRACKING == 1 ) && ( isSC == 0 ) ) memset ( hottest_keys_pointers , 0 , sizeof ( uint16_t ) * HOTTEST_KEYS_TO_TRACK ) ; while ( i < CACHE_BATCH_SIZE && trace [ trace_iter ] . opcode != NOP ) { if ( ( ENABLE_HOT_KEY_TRACKING == 1 ) && ( DISABLE_CACHE == 0 ) ) { if ( trace [ trace_iter ] . key_id < HOTTEST_KEYS_TO_TRACK ) { if ( IS_READ ( trace [ trace_iter ] . opcode ) == 1 ) { if ( hottest_keys_pointers [ trace [ trace_iter ] . key_id ] == 0 ) { hottest_keys_pointers [ trace [ trace_iter ] . key_id ] = i + 1 ; if ( isSC == 1 ) ops [ i ] . value [ 0 ] = ( uint8_t ) trace [ trace_iter ] . key_id ; } else { uint16_t op_i = hottest_keys_pointers [ trace [ trace_iter ] . key_id ] - 1 ; if ( ENABLE_ASSERTIONS ) { assert ( ( hottest_keys_pointers [ trace [ trace_iter ] . key_id ] > 0 ) ) ; assert ( ops [ op_i ] . opcode == CACHE_OP_GET ) ; assert ( ops [ op_i ] . val_len < 255 ) ; } ops [ op_i ] . val_len ++ ; trace_iter ++ ; if ( trace [ trace_iter ] . opcode == NOP ) trace_iter = 0 ; continue ; } } else hottest_keys_pointers [ trace [ trace_iter ] . key_id ] = 0 ; } } * ( uint128 * ) & ops [ i ] = trace [ trace_iter ] . key_hash ; if ( MEASURE_LATENCY == 1 ) ops [ i ] . key . meta . state = 0 ; is_update = ( IS_WRITE ( trace [ trace_iter ] . opcode ) ) ? ( uint8_t ) 1 : ( uint8_t ) 0 ; if ( ENABLE_ASSERTIONS == 1 ) assert ( WRITE_RATIO > 0 || is_update == 0 ) ; ops [ i ] . opcode = is_update ? ( uint8_t ) CACHE_OP_PUT : ( uint8_t ) CACHE_OP_GET ; ops [ i ] . val_len = is_update ? ( uint8_t ) ( HERD_VALUE_SIZE >> SHIFT_BITS ) : ( uint8_t ) 0 ; if ( is_update ) if ( ENABLE_ASSERTIONS == 1 ) assert ( ops [ i ] . val_len > 0 ) ; ops [ i ] . key . meta . cid = ( uint8_t ) ( ( ENABLE_MULTIPLE_SESSIONS == 1 ) ? ( uint8_t ) ( hrd_fastrand ( & seed ) % SESSIONS_PER_CLIENT ) * CLIENTS_PER_MACHINE + thread_id : thread_id ) ; if ( MEASURE_LATENCY == 1 ) start_measurement ( start , latency_info , trace [ trace_iter ] . home_machine_id , ops , i , thread_id , trace [ trace_iter ] . opcode , isSC , next_op_i ) ; kh [ i ] . machine = ( uint8_t ) trace [ trace_iter ] . home_machine_id ; kh [ i ] . worker = ( uint8_t ) trace [ trace_iter ] . home_worker_id ; resp [ i ] . type = EMPTY ; if ( ops [ i ] . opcode == CACHE_OP_PUT ) str_to_binary ( ops [ i ] . value , " " , HERD_VALUE_SIZE ) ; trace_iter ++ ; empty_reqs ++ ; if ( DISABLE_CACHE == 1 ) resp [ i ] . type = CACHE_MISS ; i ++ ; if ( trace [ trace_iter ] . opcode == NOP ) trace_iter = 0 ; } empty_reqs *= 100 ; c_stats [ thread_id ] . empty_reqs_per_trace = ( double ) empty_reqs / CACHE_BATCH_SIZE ; c_stats [ thread_id ] . tot_empty_reqs_per_trace += c_stats [ thread_id ] . empty_reqs_per_trace ; if ( DISABLE_CACHE == 0 ) { if ( isSC == 1 ) cache_batch_op_lin_non_stalling_sessions ( i , thread_id , & ops , resp ) ; else cache_batch_op_sc ( i , thread_id , & ops , resp ) ; } if ( MEASURE_LATENCY == 1 && ( DISABLE_CACHE == 0 ) ) { hot_request_bookkeeping_for_latency_measurements ( start , latency_info , ops , i , thread_id , isSC , resp ) ; } if ( ENABLE_ASSERTIONS == 1 ) { for ( j = 0 ; j < CACHE_BATCH_SIZE ; j ++ ) { assert ( resp [ j ] . type != RETRY ) ; } } return trace_iter ; }
refill empty slots of array from the trace
refill empty slots of array from the trace
//  green_printf("NEW BUFFER \n"); //tag the op_i+1 such that '0' means unused //the op tracks the key it stores //          yellow_printf("Start tracking key %d , in op  %d\n", trace[trace_iter].key_id, i); //            printf("opcode %d, op_i %d key %d key_pointer_to_op %d \n", //                   ops[op_i].opcode, op_i, trace[trace_iter].key_id, hottest_keys_pointers[trace[trace_iter].key_id]); //          yellow_printf("tracked key %d , in op  %d, new val_len %d\n", trace[trace_iter].key_id, op_i, ops[op_i].val_len); // on a write // *(uint128 *) &ops[i] = CityHash128((char *) &(trace[trace_iter].key_id), 4); // printf("Cache: Key = %d, tag: %d. \n", trace[trace_iter].key_id, ops[i].key.tag); // if it is not an update the val_len will not be included in the packet // only for multiple sessions //put the folowing value