BurstFS_Server/burstfs_request_manager.c
int rm_receive_remote_message(int app_id,\
		int sock_id, long tot_sz){

	int rc = ULFS_SUCCESS;

	long dbg_tot_recv = 0;
	app_config_t *app_config =\
			(app_config_t *)arraylist_get(app_config_list, app_id);
	int thrd_id = app_config->thrd_idxs[sock_id];
	thrd_ctrl_t *thrd_ctrl =\
			(thrd_ctrl_t *)arraylist_get(thrd_list, thrd_id);


    int irecv_flag[RECV_BUF_CNT] = {0};
    MPI_Request recv_req[RECV_BUF_CNT] = {0};
    MPI_Status status;

    /*
     * ToDo: something wrong happens and tot_sz keeps larger
     * than 0, handle this exception.
     * */

	shm_wait_tm.tv_sec = 0;
	shm_wait_tm.tv_nsec = SHM_WAIT_INTERVAL;

    while (tot_sz > 0) {
		 int i, return_code;

		 for (i = 0; i < RECV_BUF_CNT; i++) {
			 MPI_Irecv(thrd_ctrl->del_recv_msg_buf[i], RECV_BUF_LEN,\
					 MPI_CHAR, MPI_ANY_SOURCE,\
					 	 SER_DATA_TAG + thrd_id,\
						 	 MPI_COMM_WORLD, &recv_req[i]);
		 }

		 int recv_counter = 0;
		 while(tot_sz > 0) {
			 for (i = 0; i < RECV_BUF_CNT; i++) {
					if (irecv_flag[i] == 0) {
						return_code = MPI_Test(&recv_req[i],\
								&irecv_flag[i], &status);
						if (return_code != MPI_SUCCESS) {
							return ULFS_ERROR_RM_RECV;
						}

						if (irecv_flag[i] != 0) {

							rc = rm_process_received_msg(app_id, \
									sock_id, thrd_ctrl->del_recv_msg_buf[i],\
									&tot_sz);
							if (rc != ULFS_SUCCESS)
								return rc;
							recv_counter++;
							dbg_tot_recv++;

						}
					}
			 }

			 if (recv_counter == RECV_BUF_CNT) {
					for (i = 0; i < RECV_BUF_CNT; i++) {
						irecv_flag[i] = 0;
					}
					recv_counter = 0;
					break;
			 }

		 }

	 }

	/*purify shared receive buffer*/
	rc = sock_notify_cli(sock_id, COMM_DIGEST);
	if (rc != 0) {
		return rc;
	}

	int client_id = app_config->client_ranks[sock_id];

	int *ptr_size = (int *)app_config->shm_recv_bufs[client_id];
	while (*ptr_size != 0) {
		nanosleep(&shm_wait_tm, NULL);
	}

	 return ULFS_SUCCESS;
}
/**
*
* receive the requested data returned as a result of
* the delegated read requests
* @param app_id:
* @param sock_id:
* @param tot_sz: the total data size to receive
* @return success/error code
*/
receive the requested data returned as a result of
the delegated read requests
receive the requested data returned as a result of the delegated read requests
int rm_receive_remote_message(int app_id,\
		int sock_id, long tot_sz){
	int rc = ULFS_SUCCESS;
	long dbg_tot_recv = 0;
	app_config_t *app_config =\
			(app_config_t *)arraylist_get(app_config_list, app_id);
	int thrd_id = app_config->thrd_idxs[sock_id];
	thrd_ctrl_t *thrd_ctrl =\
			(thrd_ctrl_t *)arraylist_get(thrd_list, thrd_id);
    int irecv_flag[RECV_BUF_CNT] = {0};
    MPI_Request recv_req[RECV_BUF_CNT] = {0};
    MPI_Status status;
	shm_wait_tm.tv_sec = 0;
	shm_wait_tm.tv_nsec = SHM_WAIT_INTERVAL;
    while (tot_sz > 0) {
		 int i, return_code;
		 for (i = 0; i < RECV_BUF_CNT; i++) {
			 MPI_Irecv(thrd_ctrl->del_recv_msg_buf[i], RECV_BUF_LEN,\
					 MPI_CHAR, MPI_ANY_SOURCE,\
					 	 SER_DATA_TAG + thrd_id,\
						 	 MPI_COMM_WORLD, &recv_req[i]);
		 }
		 int recv_counter = 0;
		 while(tot_sz > 0) {
			 for (i = 0; i < RECV_BUF_CNT; i++) {
					if (irecv_flag[i] == 0) {
						return_code = MPI_Test(&recv_req[i],\
								&irecv_flag[i], &status);
						if (return_code != MPI_SUCCESS) {
							return ULFS_ERROR_RM_RECV;
						}
						if (irecv_flag[i] != 0) {
							rc = rm_process_received_msg(app_id, \
									sock_id, thrd_ctrl->del_recv_msg_buf[i],\
									&tot_sz);
							if (rc != ULFS_SUCCESS)
								return rc;
							recv_counter++;
							dbg_tot_recv++;
						}
					}
			 }
			 if (recv_counter == RECV_BUF_CNT) {
					for (i = 0; i < RECV_BUF_CNT; i++) {
						irecv_flag[i] = 0;
					}
					recv_counter = 0;
					break;
			 }
		 }
	 }
	rc = sock_notify_cli(sock_id, COMM_DIGEST);
	if (rc != 0) {
		return rc;
	}
	int client_id = app_config->client_ranks[sock_id];
	int *ptr_size = (int *)app_config->shm_recv_bufs[client_id];
	while (*ptr_size != 0) {
		nanosleep(&shm_wait_tm, NULL);
	}
	 return ULFS_SUCCESS;
}
int rm_receive_remote_message ( int app_id , int sock_id , long tot_sz ) { int rc = ULFS_SUCCESS ; long dbg_tot_recv = 0 ; app_config_t * app_config = ( app_config_t * ) arraylist_get ( app_config_list , app_id ) ; int thrd_id = app_config -> thrd_idxs [ sock_id ] ; thrd_ctrl_t * thrd_ctrl = ( thrd_ctrl_t * ) arraylist_get ( thrd_list , thrd_id ) ; int irecv_flag [ RECV_BUF_CNT ] = { 0 } ; MPI_Request recv_req [ RECV_BUF_CNT ] = { 0 } ; MPI_Status status ; shm_wait_tm . tv_sec = 0 ; shm_wait_tm . tv_nsec = SHM_WAIT_INTERVAL ; while ( tot_sz > 0 ) { int i , return_code ; for ( i = 0 ; i < RECV_BUF_CNT ; i ++ ) { MPI_Irecv ( thrd_ctrl -> del_recv_msg_buf [ i ] , RECV_BUF_LEN , MPI_CHAR , MPI_ANY_SOURCE , SER_DATA_TAG + thrd_id , MPI_COMM_WORLD , & recv_req [ i ] ) ; } int recv_counter = 0 ; while ( tot_sz > 0 ) { for ( i = 0 ; i < RECV_BUF_CNT ; i ++ ) { if ( irecv_flag [ i ] == 0 ) { return_code = MPI_Test ( & recv_req [ i ] , & irecv_flag [ i ] , & status ) ; if ( return_code != MPI_SUCCESS ) { return ULFS_ERROR_RM_RECV ; } if ( irecv_flag [ i ] != 0 ) { rc = rm_process_received_msg ( app_id , sock_id , thrd_ctrl -> del_recv_msg_buf [ i ] , & tot_sz ) ; if ( rc != ULFS_SUCCESS ) return rc ; recv_counter ++ ; dbg_tot_recv ++ ; } } } if ( recv_counter == RECV_BUF_CNT ) { for ( i = 0 ; i < RECV_BUF_CNT ; i ++ ) { irecv_flag [ i ] = 0 ; } recv_counter = 0 ; break ; } } } rc = sock_notify_cli ( sock_id , COMM_DIGEST ) ; if ( rc != 0 ) { return rc ; } int client_id = app_config -> client_ranks [ sock_id ] ; int * ptr_size = ( int * ) app_config -> shm_recv_bufs [ client_id ] ; while ( * ptr_size != 0 ) { nanosleep ( & shm_wait_tm , NULL ) ; } return ULFS_SUCCESS ; }
receive the requested data returned as a result of
the delegated read requests
receive the requested data returned as a result of the delegated read requests
/*
     * ToDo: something wrong happens and tot_sz keeps larger
     * than 0, handle this exception.
     * */ /*purify shared receive buffer*/