BurstFS_Server/burstfs_cmd_handler.c
int sync_with_client(char *cmd_buf, int sock_id) {
	int app_id = *((int *)(cmd_buf)+1);
	int local_rank_idx = *((int *)(cmd_buf) + 2);
	int dbg_rank = *((int *)(cmd_buf) + 3);

	app_config_t *tmp_config;
	/*if this client is from a new application, then
	 * initialize this application's information
	 * */

	int rc;
	if (arraylist_get(app_config_list, app_id) == NULL) {
			int num_procs_per_node = *((int *)(cmd_buf) + 4);
			int req_buf_sz = *((int *)(cmd_buf) + 5);
			int recv_buf_sz = *((int *)(cmd_buf) + 6);
			long superblock_sz =\
					*((long *)(cmd_buf +\
							7 * sizeof(int)));
			long meta_offset = *((long *)(cmd_buf + 7 * sizeof(int)\
					+ sizeof(long)));
			long meta_size = *((long *)(cmd_buf + 7 * sizeof(int)\
					+ 2 * sizeof(long)));

			long fmeta_offset = *((long *)(cmd_buf + 7 * sizeof(int)\
					+ 3 * sizeof(long)));
			long fmeta_size = *((long *)(cmd_buf + 7 * sizeof(int)\
					+ 4 * sizeof(long)));

			long data_offset = *((long *)(cmd_buf + 7 * sizeof(int)\
					+ 5 * sizeof(long)));
			long data_size = *((long *)(cmd_buf + 7 * sizeof(int)\
					+ 6 * sizeof(long)));

			int cursor = 7 * sizeof(int)\
					+ 7 * sizeof(long);

/*			LOG(LOG_DBG, "superblock_sz:%ld, num_procs_per_node:%ld,\
					 req_buf_sz:%ld, data_size:%ld\n", \
					superblock_sz, num_procs_per_node, req_buf_sz, data_size); */
			tmp_config = (app_config_t *)malloc(sizeof(app_config_t));
			memcpy(tmp_config->external_spill_dir,\
					cmd_buf + cursor, MAX_PATH_LEN);

			/*don't forget to free*/
			tmp_config->num_procs_per_node = num_procs_per_node;
			tmp_config->req_buf_sz = req_buf_sz;
			tmp_config->recv_buf_sz = recv_buf_sz;
			tmp_config->superblock_sz = superblock_sz;

			tmp_config->meta_offset = meta_offset;
			tmp_config->meta_size = meta_size;

			tmp_config->fmeta_offset = fmeta_offset;
			tmp_config->fmeta_size = fmeta_size;

			tmp_config->data_offset = data_offset;
			tmp_config->data_size = data_size;

			int i;
			for (i = 0; i < MAX_NUM_CLIENTS; i++) {
				tmp_config->client_ranks[i] = -1;
				tmp_config->shm_recv_bufs[i] = NULL;
				tmp_config->shm_req_bufs[i] = NULL;
				tmp_config->shm_superblocks[i] = NULL;
				tmp_config->shm_superblock_fds[i] = -1;
				tmp_config->shm_recv_fds[i] = -1;
				tmp_config->shm_req_fds[i] = -1;
				tmp_config->spill_log_fds[i] = -1;
				tmp_config->spill_index_log_fds[i] = -1;
			}

			rc = arraylist_insert(app_config_list, app_id, tmp_config);
			if (rc != 0)
				return rc;
	} else {
		tmp_config = (app_config_t *)arraylist_get(app_config_list,\
				app_id);
	}
		/* The following code attach a delegator thread
		 * to this new connection */
		thrd_ctrl_t *thrd_ctrl =\
				(thrd_ctrl_t *)malloc(sizeof(thrd_ctrl_t));
		memset(thrd_ctrl, 0, sizeof(thrd_ctrl_t));

		thrd_ctrl->exit_flag = 0;
		cli_signature_t *cli_signature =\
				(cli_signature_t *)malloc(sizeof(cli_signature_t));
		cli_signature->app_id = app_id;
		cli_signature->sock_id = sock_id;
		rc = pthread_mutex_init(&(thrd_ctrl->thrd_lock), NULL);
		if (rc != 0) {
			return ULFS_ERROR_THRDINIT;
		}

		rc = pthread_cond_init(&(thrd_ctrl->thrd_cond), NULL);
		if (rc != 0) {
			return ULFS_ERROR_THRDINIT;
		}

		thrd_ctrl->del_req_set =\
				(msg_meta_t *)malloc(sizeof(msg_meta_t));
		if (!thrd_ctrl->del_req_set) {
			return ULFS_ERROR_NOMEM;
		}
		memset(thrd_ctrl->del_req_set,\
				0, sizeof(msg_meta_t));

		thrd_ctrl->del_req_stat =\
				(del_req_stat_t *)malloc(sizeof(del_req_stat_t));
		if (!thrd_ctrl->del_req_stat)
			return ULFS_ERROR_NOMEM;
		memset(thrd_ctrl->del_req_stat, 0, sizeof(del_req_stat_t));

		thrd_ctrl->del_req_stat->req_stat =\
				(per_del_stat_t *)malloc(sizeof(per_del_stat_t) * glb_size);
		if (!thrd_ctrl->del_req_stat->req_stat)
			return ULFS_ERROR_NOMEM;
		memset(thrd_ctrl->del_req_stat->req_stat,\
				0, sizeof(per_del_stat_t) * glb_size);
		rc = arraylist_add(thrd_list, thrd_ctrl);
		if (rc != 0)
			return rc;

		tmp_config->thrd_idxs[sock_id] = arraylist_size(thrd_list) - 1;
		tmp_config->client_ranks[sock_id] = local_rank_idx;
		tmp_config->dbg_ranks[sock_id] = dbg_rank; /*add debug rank*/

		invert_sock_ids[sock_id] = app_id;

		rc = attach_to_shm(tmp_config, app_id, sock_id);
		if (rc != ULFS_SUCCESS)
			return rc;

		rc = open_log_file(tmp_config, app_id, sock_id);
		if (rc < 0)
			return rc;

		thrd_ctrl->has_waiting_delegator = 0;
		thrd_ctrl->has_waiting_dispatcher = 0;
		rc = pthread_create(&(thrd_ctrl->thrd),\
				NULL, rm_delegate_request, (void *)(cli_signature));
		if (rc != 0) {
			return  ULFS_ERROR_THRDINIT;
		}

		return rc;
}
/**
* receive and store the client-side information,
* then attach to the client-side shared buffers.
*
* @param cmd_buf: received command from client
* @param sock_id: position in poll_set
* @return success/error code
*/
receive and store the client-side information,
then attach to the client-side shared buffers.
receive and store the client - side information then attach to the client - side shared buffers .
int sync_with_client(char *cmd_buf, int sock_id) {
	int app_id = *((int *)(cmd_buf)+1);
	int local_rank_idx = *((int *)(cmd_buf) + 2);
	int dbg_rank = *((int *)(cmd_buf) + 3);
	app_config_t *tmp_config;
	int rc;
	if (arraylist_get(app_config_list, app_id) == NULL) {
			int num_procs_per_node = *((int *)(cmd_buf) + 4);
			int req_buf_sz = *((int *)(cmd_buf) + 5);
			int recv_buf_sz = *((int *)(cmd_buf) + 6);
			long superblock_sz =\
					*((long *)(cmd_buf +\
							7 * sizeof(int)));
			long meta_offset = *((long *)(cmd_buf + 7 * sizeof(int)\
					+ sizeof(long)));
			long meta_size = *((long *)(cmd_buf + 7 * sizeof(int)\
					+ 2 * sizeof(long)));
			long fmeta_offset = *((long *)(cmd_buf + 7 * sizeof(int)\
					+ 3 * sizeof(long)));
			long fmeta_size = *((long *)(cmd_buf + 7 * sizeof(int)\
					+ 4 * sizeof(long)));
			long data_offset = *((long *)(cmd_buf + 7 * sizeof(int)\
					+ 5 * sizeof(long)));
			long data_size = *((long *)(cmd_buf + 7 * sizeof(int)\
					+ 6 * sizeof(long)));
			int cursor = 7 * sizeof(int)\
					+ 7 * sizeof(long);
			tmp_config = (app_config_t *)malloc(sizeof(app_config_t));
			memcpy(tmp_config->external_spill_dir,\
					cmd_buf + cursor, MAX_PATH_LEN);
			tmp_config->num_procs_per_node = num_procs_per_node;
			tmp_config->req_buf_sz = req_buf_sz;
			tmp_config->recv_buf_sz = recv_buf_sz;
			tmp_config->superblock_sz = superblock_sz;
			tmp_config->meta_offset = meta_offset;
			tmp_config->meta_size = meta_size;
			tmp_config->fmeta_offset = fmeta_offset;
			tmp_config->fmeta_size = fmeta_size;
			tmp_config->data_offset = data_offset;
			tmp_config->data_size = data_size;
			int i;
			for (i = 0; i < MAX_NUM_CLIENTS; i++) {
				tmp_config->client_ranks[i] = -1;
				tmp_config->shm_recv_bufs[i] = NULL;
				tmp_config->shm_req_bufs[i] = NULL;
				tmp_config->shm_superblocks[i] = NULL;
				tmp_config->shm_superblock_fds[i] = -1;
				tmp_config->shm_recv_fds[i] = -1;
				tmp_config->shm_req_fds[i] = -1;
				tmp_config->spill_log_fds[i] = -1;
				tmp_config->spill_index_log_fds[i] = -1;
			}
			rc = arraylist_insert(app_config_list, app_id, tmp_config);
			if (rc != 0)
				return rc;
	} else {
		tmp_config = (app_config_t *)arraylist_get(app_config_list,\
				app_id);
	}
		thrd_ctrl_t *thrd_ctrl =\
				(thrd_ctrl_t *)malloc(sizeof(thrd_ctrl_t));
		memset(thrd_ctrl, 0, sizeof(thrd_ctrl_t));
		thrd_ctrl->exit_flag = 0;
		cli_signature_t *cli_signature =\
				(cli_signature_t *)malloc(sizeof(cli_signature_t));
		cli_signature->app_id = app_id;
		cli_signature->sock_id = sock_id;
		rc = pthread_mutex_init(&(thrd_ctrl->thrd_lock), NULL);
		if (rc != 0) {
			return ULFS_ERROR_THRDINIT;
		}
		rc = pthread_cond_init(&(thrd_ctrl->thrd_cond), NULL);
		if (rc != 0) {
			return ULFS_ERROR_THRDINIT;
		}
		thrd_ctrl->del_req_set =\
				(msg_meta_t *)malloc(sizeof(msg_meta_t));
		if (!thrd_ctrl->del_req_set) {
			return ULFS_ERROR_NOMEM;
		}
		memset(thrd_ctrl->del_req_set,\
				0, sizeof(msg_meta_t));
		thrd_ctrl->del_req_stat =\
				(del_req_stat_t *)malloc(sizeof(del_req_stat_t));
		if (!thrd_ctrl->del_req_stat)
			return ULFS_ERROR_NOMEM;
		memset(thrd_ctrl->del_req_stat, 0, sizeof(del_req_stat_t));
		thrd_ctrl->del_req_stat->req_stat =\
				(per_del_stat_t *)malloc(sizeof(per_del_stat_t) * glb_size);
		if (!thrd_ctrl->del_req_stat->req_stat)
			return ULFS_ERROR_NOMEM;
		memset(thrd_ctrl->del_req_stat->req_stat,\
				0, sizeof(per_del_stat_t) * glb_size);
		rc = arraylist_add(thrd_list, thrd_ctrl);
		if (rc != 0)
			return rc;
		tmp_config->thrd_idxs[sock_id] = arraylist_size(thrd_list) - 1;
		tmp_config->client_ranks[sock_id] = local_rank_idx;
		tmp_config->dbg_ranks[sock_id] = dbg_rank; 
		invert_sock_ids[sock_id] = app_id;
		rc = attach_to_shm(tmp_config, app_id, sock_id);
		if (rc != ULFS_SUCCESS)
			return rc;
		rc = open_log_file(tmp_config, app_id, sock_id);
		if (rc < 0)
			return rc;
		thrd_ctrl->has_waiting_delegator = 0;
		thrd_ctrl->has_waiting_dispatcher = 0;
		rc = pthread_create(&(thrd_ctrl->thrd),\
				NULL, rm_delegate_request, (void *)(cli_signature));
		if (rc != 0) {
			return  ULFS_ERROR_THRDINIT;
		}
		return rc;
}
int sync_with_client ( char * cmd_buf , int sock_id ) { int app_id = * ( ( int * ) ( cmd_buf ) + 1 ) ; int local_rank_idx = * ( ( int * ) ( cmd_buf ) + 2 ) ; int dbg_rank = * ( ( int * ) ( cmd_buf ) + 3 ) ; app_config_t * tmp_config ; int rc ; if ( arraylist_get ( app_config_list , app_id ) == NULL ) { int num_procs_per_node = * ( ( int * ) ( cmd_buf ) + 4 ) ; int req_buf_sz = * ( ( int * ) ( cmd_buf ) + 5 ) ; int recv_buf_sz = * ( ( int * ) ( cmd_buf ) + 6 ) ; long superblock_sz = * ( ( long * ) ( cmd_buf + 7 * sizeof ( int ) ) ) ; long meta_offset = * ( ( long * ) ( cmd_buf + 7 * sizeof ( int ) + sizeof ( long ) ) ) ; long meta_size = * ( ( long * ) ( cmd_buf + 7 * sizeof ( int ) + 2 * sizeof ( long ) ) ) ; long fmeta_offset = * ( ( long * ) ( cmd_buf + 7 * sizeof ( int ) + 3 * sizeof ( long ) ) ) ; long fmeta_size = * ( ( long * ) ( cmd_buf + 7 * sizeof ( int ) + 4 * sizeof ( long ) ) ) ; long data_offset = * ( ( long * ) ( cmd_buf + 7 * sizeof ( int ) + 5 * sizeof ( long ) ) ) ; long data_size = * ( ( long * ) ( cmd_buf + 7 * sizeof ( int ) + 6 * sizeof ( long ) ) ) ; int cursor = 7 * sizeof ( int ) + 7 * sizeof ( long ) ; tmp_config = ( app_config_t * ) malloc ( sizeof ( app_config_t ) ) ; memcpy ( tmp_config -> external_spill_dir , cmd_buf + cursor , MAX_PATH_LEN ) ; tmp_config -> num_procs_per_node = num_procs_per_node ; tmp_config -> req_buf_sz = req_buf_sz ; tmp_config -> recv_buf_sz = recv_buf_sz ; tmp_config -> superblock_sz = superblock_sz ; tmp_config -> meta_offset = meta_offset ; tmp_config -> meta_size = meta_size ; tmp_config -> fmeta_offset = fmeta_offset ; tmp_config -> fmeta_size = fmeta_size ; tmp_config -> data_offset = data_offset ; tmp_config -> data_size = data_size ; int i ; for ( i = 0 ; i < MAX_NUM_CLIENTS ; i ++ ) { tmp_config -> client_ranks [ i ] = -1 ; tmp_config -> shm_recv_bufs [ i ] = NULL ; tmp_config -> shm_req_bufs [ i ] = NULL ; tmp_config -> shm_superblocks [ i ] = NULL ; tmp_config -> shm_superblock_fds [ i ] = -1 ; tmp_config -> shm_recv_fds [ i ] = -1 ; tmp_config -> shm_req_fds [ i ] = -1 ; tmp_config -> spill_log_fds [ i ] = -1 ; tmp_config -> spill_index_log_fds [ i ] = -1 ; } rc = arraylist_insert ( app_config_list , app_id , tmp_config ) ; if ( rc != 0 ) return rc ; } else { tmp_config = ( app_config_t * ) arraylist_get ( app_config_list , app_id ) ; } thrd_ctrl_t * thrd_ctrl = ( thrd_ctrl_t * ) malloc ( sizeof ( thrd_ctrl_t ) ) ; memset ( thrd_ctrl , 0 , sizeof ( thrd_ctrl_t ) ) ; thrd_ctrl -> exit_flag = 0 ; cli_signature_t * cli_signature = ( cli_signature_t * ) malloc ( sizeof ( cli_signature_t ) ) ; cli_signature -> app_id = app_id ; cli_signature -> sock_id = sock_id ; rc = pthread_mutex_init ( & ( thrd_ctrl -> thrd_lock ) , NULL ) ; if ( rc != 0 ) { return ULFS_ERROR_THRDINIT ; } rc = pthread_cond_init ( & ( thrd_ctrl -> thrd_cond ) , NULL ) ; if ( rc != 0 ) { return ULFS_ERROR_THRDINIT ; } thrd_ctrl -> del_req_set = ( msg_meta_t * ) malloc ( sizeof ( msg_meta_t ) ) ; if ( ! thrd_ctrl -> del_req_set ) { return ULFS_ERROR_NOMEM ; } memset ( thrd_ctrl -> del_req_set , 0 , sizeof ( msg_meta_t ) ) ; thrd_ctrl -> del_req_stat = ( del_req_stat_t * ) malloc ( sizeof ( del_req_stat_t ) ) ; if ( ! thrd_ctrl -> del_req_stat ) return ULFS_ERROR_NOMEM ; memset ( thrd_ctrl -> del_req_stat , 0 , sizeof ( del_req_stat_t ) ) ; thrd_ctrl -> del_req_stat -> req_stat = ( per_del_stat_t * ) malloc ( sizeof ( per_del_stat_t ) * glb_size ) ; if ( ! thrd_ctrl -> del_req_stat -> req_stat ) return ULFS_ERROR_NOMEM ; memset ( thrd_ctrl -> del_req_stat -> req_stat , 0 , sizeof ( per_del_stat_t ) * glb_size ) ; rc = arraylist_add ( thrd_list , thrd_ctrl ) ; if ( rc != 0 ) return rc ; tmp_config -> thrd_idxs [ sock_id ] = arraylist_size ( thrd_list ) - 1 ; tmp_config -> client_ranks [ sock_id ] = local_rank_idx ; tmp_config -> dbg_ranks [ sock_id ] = dbg_rank ; invert_sock_ids [ sock_id ] = app_id ; rc = attach_to_shm ( tmp_config , app_id , sock_id ) ; if ( rc != ULFS_SUCCESS ) return rc ; rc = open_log_file ( tmp_config , app_id , sock_id ) ; if ( rc < 0 ) return rc ; thrd_ctrl -> has_waiting_delegator = 0 ; thrd_ctrl -> has_waiting_dispatcher = 0 ; rc = pthread_create ( & ( thrd_ctrl -> thrd ) , NULL , rm_delegate_request , ( void * ) ( cli_signature ) ) ; if ( rc != 0 ) { return ULFS_ERROR_THRDINIT ; } return rc ; }
receive and store the client-side information,
then attach to the client-side shared buffers.
receive and store the client - side information then attach to the client - side shared buffers .
/*if this client is from a new application, then
	 * initialize this application's information
	 * */ /*			LOG(LOG_DBG, "superblock_sz:%ld, num_procs_per_node:%ld,\
					 req_buf_sz:%ld, data_size:%ld\n", \
					superblock_sz, num_procs_per_node, req_buf_sz, data_size); */ /*don't forget to free*/ /* The following code attach a delegator thread
		 * to this new connection */ /*add debug rank*/