BurstFS_Server/burstfs_request_manager.c
int rm_read_remote_data(int sock_id, int req_num) {

	int rc;

	int app_id = invert_sock_ids[sock_id];
	app_config_t *app_config =\
			(app_config_t *)arraylist_get(app_config_list, app_id);

	int client_id = app_config->client_ranks[sock_id];
	int dbg_rank = app_config->dbg_ranks[sock_id];

	int thrd_id = app_config->thrd_idxs[sock_id];
	thrd_ctrl_t *thrd_ctrl = (thrd_ctrl_t *)arraylist_get(thrd_list, thrd_id);

	pthread_mutex_lock(&thrd_ctrl->thrd_lock);

	/* get the locations of all the read requests from the key-value store*/
	rc = meta_batch_get(app_id, client_id, thrd_id, dbg_rank,\
			app_config->shm_req_bufs[client_id], req_num, \
				thrd_ctrl->del_req_set);

	/*
	 * group together the read requests
	 * to be sent to the same delegators.
	 * */
	qsort(thrd_ctrl->del_req_set->msg_meta,\
			thrd_ctrl->del_req_set->num,\
				sizeof(send_msg_t), compare_delegators);
//	print_send_msgs(thrd_ctrl->del_req_set->msg_meta, \
			thrd_ctrl->del_req_set->num, dbg_rank);
	thrd_ctrl->del_req_stat->req_stat[0].req_cnt = 1;



	int i, del_cnt = 0;
	thrd_ctrl->del_req_stat->req_stat[0].del_id =\
			thrd_ctrl->del_req_set->msg_meta[0].dest_delegator_rank;

	/* calculate the number of read requests
	 * to be sent to each delegator*/
	for (i = 1; i < thrd_ctrl->del_req_set->num; i++) {
		if (thrd_ctrl->del_req_set->msg_meta[i].dest_delegator_rank\
				== thrd_ctrl->del_req_set->msg_meta[i - 1].dest_delegator_rank) {
			thrd_ctrl->del_req_stat->req_stat[del_cnt].req_cnt++;
		} else {
			del_cnt++;
			thrd_ctrl->del_req_stat->req_stat[del_cnt].req_cnt = 1;
			thrd_ctrl->del_req_stat->req_stat[del_cnt].del_id = \
					thrd_ctrl->del_req_set->msg_meta[i].dest_delegator_rank;

		}
	}
	del_cnt++;

	thrd_ctrl->del_req_stat->del_cnt = del_cnt;

	// print_remote_del_reqs(app_id, client_id, dbg_rank, \
			thrd_ctrl->del_req_stat);

	/*wake up the service thread for the requesting client*/
	if (!thrd_ctrl->has_waiting_delegator) {
		thrd_ctrl->has_waiting_dispatcher = 1;
		pthread_cond_wait(&thrd_ctrl->thrd_cond, &thrd_ctrl->thrd_lock);
		thrd_ctrl->has_waiting_dispatcher = 0;
		pthread_cond_signal(&thrd_ctrl->thrd_cond);
	} else {
		pthread_cond_signal(&thrd_ctrl->thrd_cond);
	}
	pthread_mutex_unlock(&thrd_ctrl->thrd_lock);
	return rc;

}
/**
* send the read requests to the
* remote delegators
* @param sock_id: which socket in the poll_set received
* the application's requests
* @param req_num: number of read requests
* @return success/error code
*/
send the read requests to the
remote delegators
send the read requests to the remote delegators
int rm_read_remote_data(int sock_id, int req_num) {
	int rc;
	int app_id = invert_sock_ids[sock_id];
	app_config_t *app_config =\
			(app_config_t *)arraylist_get(app_config_list, app_id);
	int client_id = app_config->client_ranks[sock_id];
	int dbg_rank = app_config->dbg_ranks[sock_id];
	int thrd_id = app_config->thrd_idxs[sock_id];
	thrd_ctrl_t *thrd_ctrl = (thrd_ctrl_t *)arraylist_get(thrd_list, thrd_id);
	pthread_mutex_lock(&thrd_ctrl->thrd_lock);
	rc = meta_batch_get(app_id, client_id, thrd_id, dbg_rank,\
			app_config->shm_req_bufs[client_id], req_num, \
				thrd_ctrl->del_req_set);
	qsort(thrd_ctrl->del_req_set->msg_meta,\
			thrd_ctrl->del_req_set->num,\
				sizeof(send_msg_t), compare_delegators);
	thrd_ctrl->del_req_stat->req_stat[0].req_cnt = 1;
	int i, del_cnt = 0;
	thrd_ctrl->del_req_stat->req_stat[0].del_id =\
			thrd_ctrl->del_req_set->msg_meta[0].dest_delegator_rank;
	for (i = 1; i < thrd_ctrl->del_req_set->num; i++) {
		if (thrd_ctrl->del_req_set->msg_meta[i].dest_delegator_rank\
				== thrd_ctrl->del_req_set->msg_meta[i - 1].dest_delegator_rank) {
			thrd_ctrl->del_req_stat->req_stat[del_cnt].req_cnt++;
		} else {
			del_cnt++;
			thrd_ctrl->del_req_stat->req_stat[del_cnt].req_cnt = 1;
			thrd_ctrl->del_req_stat->req_stat[del_cnt].del_id = \
					thrd_ctrl->del_req_set->msg_meta[i].dest_delegator_rank;
		}
	}
	del_cnt++;
	thrd_ctrl->del_req_stat->del_cnt = del_cnt;
	if (!thrd_ctrl->has_waiting_delegator) {
		thrd_ctrl->has_waiting_dispatcher = 1;
		pthread_cond_wait(&thrd_ctrl->thrd_cond, &thrd_ctrl->thrd_lock);
		thrd_ctrl->has_waiting_dispatcher = 0;
		pthread_cond_signal(&thrd_ctrl->thrd_cond);
	} else {
		pthread_cond_signal(&thrd_ctrl->thrd_cond);
	}
	pthread_mutex_unlock(&thrd_ctrl->thrd_lock);
	return rc;
}
int rm_read_remote_data ( int sock_id , int req_num ) { int rc ; int app_id = invert_sock_ids [ sock_id ] ; app_config_t * app_config = ( app_config_t * ) arraylist_get ( app_config_list , app_id ) ; int client_id = app_config -> client_ranks [ sock_id ] ; int dbg_rank = app_config -> dbg_ranks [ sock_id ] ; int thrd_id = app_config -> thrd_idxs [ sock_id ] ; thrd_ctrl_t * thrd_ctrl = ( thrd_ctrl_t * ) arraylist_get ( thrd_list , thrd_id ) ; pthread_mutex_lock ( & thrd_ctrl -> thrd_lock ) ; rc = meta_batch_get ( app_id , client_id , thrd_id , dbg_rank , app_config -> shm_req_bufs [ client_id ] , req_num , thrd_ctrl -> del_req_set ) ; qsort ( thrd_ctrl -> del_req_set -> msg_meta , thrd_ctrl -> del_req_set -> num , sizeof ( send_msg_t ) , compare_delegators ) ; thrd_ctrl -> del_req_stat -> req_stat [ 0 ] . req_cnt = 1 ; int i , del_cnt = 0 ; thrd_ctrl -> del_req_stat -> req_stat [ 0 ] . del_id = thrd_ctrl -> del_req_set -> msg_meta [ 0 ] . dest_delegator_rank ; for ( i = 1 ; i < thrd_ctrl -> del_req_set -> num ; i ++ ) { if ( thrd_ctrl -> del_req_set -> msg_meta [ i ] . dest_delegator_rank == thrd_ctrl -> del_req_set -> msg_meta [ i - 1 ] . dest_delegator_rank ) { thrd_ctrl -> del_req_stat -> req_stat [ del_cnt ] . req_cnt ++ ; } else { del_cnt ++ ; thrd_ctrl -> del_req_stat -> req_stat [ del_cnt ] . req_cnt = 1 ; thrd_ctrl -> del_req_stat -> req_stat [ del_cnt ] . del_id = thrd_ctrl -> del_req_set -> msg_meta [ i ] . dest_delegator_rank ; } } del_cnt ++ ; thrd_ctrl -> del_req_stat -> del_cnt = del_cnt ; if ( ! thrd_ctrl -> has_waiting_delegator ) { thrd_ctrl -> has_waiting_dispatcher = 1 ; pthread_cond_wait ( & thrd_ctrl -> thrd_cond , & thrd_ctrl -> thrd_lock ) ; thrd_ctrl -> has_waiting_dispatcher = 0 ; pthread_cond_signal ( & thrd_ctrl -> thrd_cond ) ; } else { pthread_cond_signal ( & thrd_ctrl -> thrd_cond ) ; } pthread_mutex_unlock ( & thrd_ctrl -> thrd_lock ) ; return rc ; }
send the read requests to the
remote delegators
send the read requests to the remote delegators
/* get the locations of all the read requests from the key-value store*/ /*
	 * group together the read requests
	 * to be sent to the same delegators.
	 * */ //	print_send_msgs(thrd_ctrl->del_req_set->msg_meta, \
			thrd_ctrl->del_req_set->num, dbg_rank);
 /* calculate the number of read requests
	 * to be sent to each delegator*/ // print_remote_del_reqs(app_id, client_id, dbg_rank, \
			thrd_ctrl->del_req_stat);
 /*wake up the service thread for the requesting client*/