BurstFS_Server/burstfs_request_manager.c
int rm_process_received_msg(int app_id, int sock_id, \
		char *recv_msg_buf, long *ptr_tot_sz) {

	int rc;
	app_config_t *app_config =\
			(app_config_t *)arraylist_get(app_config_list, app_id);
	int client_id = app_config->client_ranks[sock_id];

	/*
	 * format of recv_msg_buf: num,
	 * src_app_id, src_cli_id,
	 * src_fid, src_offset, src_length
	 */
	int num = *(int *)recv_msg_buf;
	int recv_cursor = 0;

	/*
	 * ptr_size and ptr_num point to the
	 * size and num information of the
	 * client-side shared memory
	 * */
	int *ptr_size = NULL;
	int *ptr_num = NULL;

	recv_msg_t *tmp_recv_msg =\
		 (recv_msg_t *)(recv_msg_buf + sizeof(int));
    recv_cursor += sizeof(int);

	shm_meta_t *tmp_sh_msg;
	ptr_size =\
			(int *)app_config->shm_recv_bufs[client_id];
	ptr_num =\
			(int *)(app_config->shm_recv_bufs[client_id]\
		 + sizeof(int));

	int j;
	for (j = 0; j < num; j++) {

		if (*ptr_size + tmp_recv_msg->length + sizeof(shm_meta_t)\
				+ 2 * sizeof(int) > app_config->recv_buf_sz) {
			/*client-side receive buffer is full,
			 * wait until the client reads all the
			 * data*/
			rc = sock_notify_cli(sock_id, COMM_READ);

			if (rc != 0) {
				return rc;
			}
			while (*ptr_size != 0) {
				nanosleep(&shm_wait_tm, NULL);
			}/*wait until client digest the data*/
		}

		/*fill the next message in the shared buffer*/
		tmp_sh_msg =\
			 (shm_meta_t *)(((char *)app_config->shm_recv_bufs[client_id]\
				+ *ptr_size) + 2 * sizeof(int));

		tmp_sh_msg->src_fid = tmp_recv_msg->src_fid;
		tmp_sh_msg->offset = tmp_recv_msg->src_offset;
		tmp_sh_msg->length = tmp_recv_msg->length;

	    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];
	    // print_recv_msg(app_id, client_id, dbg_rank, \
					thrd_id, tmp_sh_msg);


		recv_cursor += sizeof(recv_msg_t);
		*ptr_size += sizeof(shm_meta_t);

		memcpy(2 * sizeof(int) \
			+ app_config->shm_recv_bufs[client_id] + *ptr_size,\
				recv_msg_buf + recv_cursor,\
					 tmp_recv_msg->length);

		*ptr_tot_sz -= tmp_recv_msg->length;
		recv_cursor += tmp_recv_msg->length;
		*ptr_size = tmp_recv_msg->length + *ptr_size;
		(*ptr_num)++;

		/*the message buffer may contain a list of messages*/
		tmp_recv_msg = (recv_msg_t *)(recv_msg_buf\
				+ recv_cursor);

	}

	return ULFS_SUCCESS;

}
/**
*
* parse the received message, and deliver to the
* client
* @param app_id: client's application id
* @param sock_id: socket index in the poll_set
* for that client
* @param recv_msg_buf: buffer for received message
*  packed read requests
* @param ptr_tot_sz: total data size to receive
* @return success/error code
*/
parse the received message, and deliver to the
client
parse the received message and deliver to the client
int rm_process_received_msg(int app_id, int sock_id, \
		char *recv_msg_buf, long *ptr_tot_sz) {
	int rc;
	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 num = *(int *)recv_msg_buf;
	int recv_cursor = 0;
	int *ptr_size = NULL;
	int *ptr_num = NULL;
	recv_msg_t *tmp_recv_msg =\
		 (recv_msg_t *)(recv_msg_buf + sizeof(int));
    recv_cursor += sizeof(int);
	shm_meta_t *tmp_sh_msg;
	ptr_size =\
			(int *)app_config->shm_recv_bufs[client_id];
	ptr_num =\
			(int *)(app_config->shm_recv_bufs[client_id]\
		 + sizeof(int));
	int j;
	for (j = 0; j < num; j++) {
		if (*ptr_size + tmp_recv_msg->length + sizeof(shm_meta_t)\
				+ 2 * sizeof(int) > app_config->recv_buf_sz) {
			rc = sock_notify_cli(sock_id, COMM_READ);
			if (rc != 0) {
				return rc;
			}
			while (*ptr_size != 0) {
				nanosleep(&shm_wait_tm, NULL);
			}
		}
		tmp_sh_msg =\
			 (shm_meta_t *)(((char *)app_config->shm_recv_bufs[client_id]\
				+ *ptr_size) + 2 * sizeof(int));
		tmp_sh_msg->src_fid = tmp_recv_msg->src_fid;
		tmp_sh_msg->offset = tmp_recv_msg->src_offset;
		tmp_sh_msg->length = tmp_recv_msg->length;
	    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];
		recv_cursor += sizeof(recv_msg_t);
		*ptr_size += sizeof(shm_meta_t);
		memcpy(2 * sizeof(int) \
			+ app_config->shm_recv_bufs[client_id] + *ptr_size,\
				recv_msg_buf + recv_cursor,\
					 tmp_recv_msg->length);
		*ptr_tot_sz -= tmp_recv_msg->length;
		recv_cursor += tmp_recv_msg->length;
		*ptr_size = tmp_recv_msg->length + *ptr_size;
		(*ptr_num)++;
		tmp_recv_msg = (recv_msg_t *)(recv_msg_buf\
				+ recv_cursor);
	}
	return ULFS_SUCCESS;
}
int rm_process_received_msg ( int app_id , int sock_id , char * recv_msg_buf , long * ptr_tot_sz ) { int rc ; 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 num = * ( int * ) recv_msg_buf ; int recv_cursor = 0 ; int * ptr_size = NULL ; int * ptr_num = NULL ; recv_msg_t * tmp_recv_msg = ( recv_msg_t * ) ( recv_msg_buf + sizeof ( int ) ) ; recv_cursor += sizeof ( int ) ; shm_meta_t * tmp_sh_msg ; ptr_size = ( int * ) app_config -> shm_recv_bufs [ client_id ] ; ptr_num = ( int * ) ( app_config -> shm_recv_bufs [ client_id ] + sizeof ( int ) ) ; int j ; for ( j = 0 ; j < num ; j ++ ) { if ( * ptr_size + tmp_recv_msg -> length + sizeof ( shm_meta_t ) + 2 * sizeof ( int ) > app_config -> recv_buf_sz ) { rc = sock_notify_cli ( sock_id , COMM_READ ) ; if ( rc != 0 ) { return rc ; } while ( * ptr_size != 0 ) { nanosleep ( & shm_wait_tm , NULL ) ; } } tmp_sh_msg = ( shm_meta_t * ) ( ( ( char * ) app_config -> shm_recv_bufs [ client_id ] + * ptr_size ) + 2 * sizeof ( int ) ) ; tmp_sh_msg -> src_fid = tmp_recv_msg -> src_fid ; tmp_sh_msg -> offset = tmp_recv_msg -> src_offset ; tmp_sh_msg -> length = tmp_recv_msg -> length ; 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 ] ; recv_cursor += sizeof ( recv_msg_t ) ; * ptr_size += sizeof ( shm_meta_t ) ; memcpy ( 2 * sizeof ( int ) + app_config -> shm_recv_bufs [ client_id ] + * ptr_size , recv_msg_buf + recv_cursor , tmp_recv_msg -> length ) ; * ptr_tot_sz -= tmp_recv_msg -> length ; recv_cursor += tmp_recv_msg -> length ; * ptr_size = tmp_recv_msg -> length + * ptr_size ; ( * ptr_num ) ++ ; tmp_recv_msg = ( recv_msg_t * ) ( recv_msg_buf + recv_cursor ) ; } return ULFS_SUCCESS ; }
parse the received message, and deliver to the
client
parse the received message and deliver to the client
/*
	 * format of recv_msg_buf: num,
	 * src_app_id, src_cli_id,
	 * src_fid, src_offset, src_length
	 */ /*
	 * ptr_size and ptr_num point to the
	 * size and num information of the
	 * client-side shared memory
	 * */ /*client-side receive buffer is full,
			 * wait until the client reads all the
			 * data*/ /*wait until client digest the data*/ /*fill the next message in the shared buffer*/ // print_recv_msg(app_id, client_id, dbg_rank, \
					thrd_id, tmp_sh_msg);
 /*the message buffer may contain a list of messages*/