include/ccKVS/inline_util.h
static inline enum control_flow_directive poll_remote_region(uint16_t* qp_i, int* pull_ptr, uint32_t* per_qp_buf_slots, struct wrkr_ud_req* req,
															 bool* multiget, uint16_t wr_i, uint16_t* received_messages, uint16_t* per_qp_received_messages,
															 struct mica_op** next_req_ptr, uint16_t* get_i, uint16_t* get_num, uint8_t* requests_per_message)
{
	uint32_t index = (pull_ptr[(*qp_i)] + 1) % per_qp_buf_slots[(*qp_i)];
	uint8_t req_opcode = req[index].m_op.opcode;
	if (req_opcode != MICA_OP_GET && req_opcode != MICA_OP_PUT) {
		if (ENABLE_COALESCING && req_opcode == MICA_OP_MULTI_GET) {
			if ((*multiget) == false) { // if it's the first request
				struct mica_op *tmp = (struct mica_op *) (req[index].m_op.value + (req[index].m_op.val_len - 2) * HERD_GET_REQ_SIZE);
				if (tmp->opcode != MICA_OP_GET) return continue_;
				(*get_num)  = req[index].m_op.val_len;
				if (wr_i + (*get_num) > WORKER_MAX_BATCH) {
					// printf("breaking %d\n", wrkr_lid);
					return break_;
				}
				(*multiget)  = true;
				requests_per_message[(*received_messages)] = *get_num;
				(*get_i)  = 0;
				req[index].m_op.opcode = MICA_OP_GET;// necessary to work with the MICA API
				(*next_req_ptr) = (struct mica_op *) &(req[index].m_op);

				(*received_messages)++;
				per_qp_received_messages[(*qp_i)]++;
                // green_printf("I see a Multi Get of %d gets \n", (*get_num) );
				if (ENABLE_ASSERTIONS && (*next_req_ptr)->opcode != MICA_OP_GET) {
					red_printf("req opcode %d, next_rq_ptr opcode %d\n",req_opcode,  (*next_req_ptr)->opcode);
				}
			}
			else if (ENABLE_ASSERTIONS) assert(false);
		}
		else { // we need to check the rest of the buffers
			if (ENABLE_ASSERTIONS)
				if(req_opcode != 0) {
					red_printf("Wrong Opcode %d \n", req_opcode);
					assert(false);
				}
			if ((*qp_i)  == WORKER_NUM_UD_QPS -1) return break_; // all qp buffers were checked
			else {
				HRD_MOD_ADD((*qp_i) , WORKER_NUM_UD_QPS);
				return continue_; // move to the next buffer
			}
		}
	}
	return no_cf_change;
}
// poll the remote request region and return a control flow directive
poll the remote request region and return a control flow directive
poll the remote request region and return a control flow directive
static inline enum control_flow_directive poll_remote_region(uint16_t* qp_i, int* pull_ptr, uint32_t* per_qp_buf_slots, struct wrkr_ud_req* req,
															 bool* multiget, uint16_t wr_i, uint16_t* received_messages, uint16_t* per_qp_received_messages,
															 struct mica_op** next_req_ptr, uint16_t* get_i, uint16_t* get_num, uint8_t* requests_per_message)
{
	uint32_t index = (pull_ptr[(*qp_i)] + 1) % per_qp_buf_slots[(*qp_i)];
	uint8_t req_opcode = req[index].m_op.opcode;
	if (req_opcode != MICA_OP_GET && req_opcode != MICA_OP_PUT) {
		if (ENABLE_COALESCING && req_opcode == MICA_OP_MULTI_GET) {
			if ((*multiget) == false) { 
				struct mica_op *tmp = (struct mica_op *) (req[index].m_op.value + (req[index].m_op.val_len - 2) * HERD_GET_REQ_SIZE);
				if (tmp->opcode != MICA_OP_GET) return continue_;
				(*get_num)  = req[index].m_op.val_len;
				if (wr_i + (*get_num) > WORKER_MAX_BATCH) {
					return break_;
				}
				(*multiget)  = true;
				requests_per_message[(*received_messages)] = *get_num;
				(*get_i)  = 0;
				req[index].m_op.opcode = MICA_OP_GET;
				(*next_req_ptr) = (struct mica_op *) &(req[index].m_op);
				(*received_messages)++;
				per_qp_received_messages[(*qp_i)]++;
				if (ENABLE_ASSERTIONS && (*next_req_ptr)->opcode != MICA_OP_GET) {
					red_printf("req opcode %d, next_rq_ptr opcode %d\n",req_opcode,  (*next_req_ptr)->opcode);
				}
			}
			else if (ENABLE_ASSERTIONS) assert(false);
		}
		else { 
			if (ENABLE_ASSERTIONS)
				if(req_opcode != 0) {
					red_printf("Wrong Opcode %d \n", req_opcode);
					assert(false);
				}
			if ((*qp_i)  == WORKER_NUM_UD_QPS -1) return break_; 
			else {
				HRD_MOD_ADD((*qp_i) , WORKER_NUM_UD_QPS);
				return continue_; 
			}
		}
	}
	return no_cf_change;
}
static inline enum control_flow_directive poll_remote_region ( uint16_t * qp_i , int * pull_ptr , uint32_t * per_qp_buf_slots , struct wrkr_ud_req * req , bool * multiget , uint16_t wr_i , uint16_t * received_messages , uint16_t * per_qp_received_messages , struct mica_op * * next_req_ptr , uint16_t * get_i , uint16_t * get_num , uint8_t * requests_per_message ) { uint32_t index = ( pull_ptr [ ( * qp_i ) ] + 1 ) % per_qp_buf_slots [ ( * qp_i ) ] ; uint8_t req_opcode = req [ index ] . m_op . opcode ; if ( req_opcode != MICA_OP_GET && req_opcode != MICA_OP_PUT ) { if ( ENABLE_COALESCING && req_opcode == MICA_OP_MULTI_GET ) { if ( ( * multiget ) == false ) { struct mica_op * tmp = ( struct mica_op * ) ( req [ index ] . m_op . value + ( req [ index ] . m_op . val_len - 2 ) * HERD_GET_REQ_SIZE ) ; if ( tmp -> opcode != MICA_OP_GET ) return continue_ ; ( * get_num ) = req [ index ] . m_op . val_len ; if ( wr_i + ( * get_num ) > WORKER_MAX_BATCH ) { return break_ ; } ( * multiget ) = true ; requests_per_message [ ( * received_messages ) ] = * get_num ; ( * get_i ) = 0 ; req [ index ] . m_op . opcode = MICA_OP_GET ; ( * next_req_ptr ) = ( struct mica_op * ) & ( req [ index ] . m_op ) ; ( * received_messages ) ++ ; per_qp_received_messages [ ( * qp_i ) ] ++ ; if ( ENABLE_ASSERTIONS && ( * next_req_ptr ) -> opcode != MICA_OP_GET ) { red_printf ( " \n " , req_opcode , ( * next_req_ptr ) -> opcode ) ; } } else if ( ENABLE_ASSERTIONS ) assert ( false ) ; } else { if ( ENABLE_ASSERTIONS ) if ( req_opcode != 0 ) { red_printf ( " \n " , req_opcode ) ; assert ( false ) ; } if ( ( * qp_i ) == WORKER_NUM_UD_QPS - 1 ) return break_ ; else { HRD_MOD_ADD ( ( * qp_i ) , WORKER_NUM_UD_QPS ) ; return continue_ ; } } } return no_cf_change ; }
poll the remote request region and return a control flow directive
poll the remote request region and return a control flow directive
// if it's the first request // printf("breaking %d\n", wrkr_lid); // necessary to work with the MICA API // green_printf("I see a Multi Get of %d gets \n", (*get_num) ); // we need to check the rest of the buffers // all qp buffers were checked // move to the next buffer