From: Pierre Veyre Date: Mon, 31 Mar 2014 14:56:45 +0000 (+0200) Subject: Improve MSG_file_read() and MSG_file_write() X-Git-Tag: v3_11~190 X-Git-Url: http://info.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/commitdiff_plain/dc45d41606e5f40070026183b0aea7bac521ef4b?ds=sidebyside Improve MSG_file_read() and MSG_file_write() --- diff --git a/include/simgrid/simix.h b/include/simgrid/simix.h index 6b68336a4f..2013686d05 100644 --- a/include/simgrid/simix.h +++ b/include/simgrid/simix.h @@ -488,8 +488,8 @@ XBT_PUBLIC(int) simcall_sem_get_capacity(smx_sem_t sem); /***************************** File **********************************/ XBT_PUBLIC(void *) simcall_file_get_data(smx_file_t fd); XBT_PUBLIC(void) simcall_file_set_data(smx_file_t fd, void *data); -XBT_PUBLIC(sg_size_t) simcall_file_read(smx_file_t fd, sg_size_t size); -XBT_PUBLIC(sg_size_t) simcall_file_write(smx_file_t fd, sg_size_t size); +XBT_PUBLIC(sg_size_t) simcall_file_read(smx_file_t fd, sg_size_t size, smx_host_t host); +XBT_PUBLIC(sg_size_t) simcall_file_write(smx_file_t fd, sg_size_t size, smx_host_t host); XBT_PUBLIC(smx_file_t) simcall_file_open(const char* fullpath); XBT_PUBLIC(int) simcall_file_close(smx_file_t fd); XBT_PUBLIC(int) simcall_file_unlink(smx_file_t fd); diff --git a/src/msg/msg_io.c b/src/msg/msg_io.c index 2c2adca044..f632e57e3d 100644 --- a/src/msg/msg_io.c +++ b/src/msg/msg_io.c @@ -6,6 +6,7 @@ #include "msg_private.h" #include "xbt/log.h" +#include "msg_mailbox.h" XBT_LOG_NEW_DEFAULT_SUBCATEGORY(msg_io, msg, "Logging specific to MSG (io)"); @@ -85,29 +86,95 @@ void MSG_file_dump (msg_file_t fd){ } /** \ingroup msg_file_management - * \brief Read a file + * \brief Read a file (local or remote) * * \param size of the file to read * \param fd is a the file descriptor - * \return the number of bytes successfully read + * \return the number of bytes successfully read or -1 if an error occurred */ sg_size_t MSG_file_read(msg_file_t fd, sg_size_t size) { - msg_file_priv_t priv = MSG_file_priv(fd); - return simcall_file_read(priv->simdata->smx_file, size); + msg_file_priv_t file_priv = MSG_file_priv(fd); + sg_size_t read_size; + + /* Find the host where the file is physically located and read it */ + msg_storage_t storage_src =(msg_storage_t) xbt_lib_get_elm_or_null(storage_lib, file_priv->storageId); + msg_storage_priv_t storage_priv_src = MSG_storage_priv(storage_src); + msg_host_t remote_host = MSG_get_host_by_name(storage_priv_src->host); + read_size = simcall_file_read(file_priv->simdata->smx_file, size, remote_host); + + if(strcmp(storage_priv_src->host, MSG_host_get_name(MSG_host_self()))){ + /* the file is hosted on a remote host, initiate a communication between src and dest hosts for data transfer */ + XBT_DEBUG("File is on %s remote host, initiate data transfer of %llu bytes.", storage_priv_src->host, read_size); + msg_host_t *m_host_list = NULL; + m_host_list = calloc(2, sizeof(msg_host_t)); + + m_host_list[0] = MSG_host_self(); + m_host_list[1] = remote_host; + double computation_amount[] = { 0, 0 }; + double communication_amount[] = { 0, (double)read_size, 0, 0 }; + + msg_task_t task = MSG_parallel_task_create("file transfer for read", 2, m_host_list, computation_amount, communication_amount, NULL); + msg_error_t transfer = MSG_parallel_task_execute(task); + MSG_task_destroy(task); + free(m_host_list); + if(transfer != MSG_OK){ + if (transfer == MSG_HOST_FAILURE) + XBT_WARN("Transfer error, remote host just turned off!"); + if (transfer == MSG_TASK_CANCELED) + XBT_WARN("Transfer error, task has been canceled!"); + + return -1; + } + } + return read_size; } /** \ingroup msg_file_management - * \brief Write into a file + * \brief Write into a file (local or remote) * * \param size of the file to write * \param fd is a the file descriptor - * \return the number of bytes successfully write + * \return the number of bytes successfully write or -1 if an error occurred */ sg_size_t MSG_file_write(msg_file_t fd, sg_size_t size) { - msg_file_priv_t priv = MSG_file_priv(fd); - return simcall_file_write(priv->simdata->smx_file, size); + msg_file_priv_t file_priv = MSG_file_priv(fd); + sg_size_t write_size; + + /* Find the host where the file is physically located (remote or local)*/ + msg_storage_t storage_src =(msg_storage_t) xbt_lib_get_elm_or_null(storage_lib, file_priv->storageId); + msg_storage_priv_t storage_priv_src = MSG_storage_priv(storage_src); + msg_host_t remote_host = MSG_get_host_by_name(storage_priv_src->host); + + if(strcmp(storage_priv_src->host, MSG_host_get_name(MSG_host_self()))){ + /* the file is hosted on a remote host, initiate a communication between src and dest hosts for data transfer */ + XBT_DEBUG("File is on %s remote host, initiate data transfer of %llu bytes.", storage_priv_src->host, size); + msg_host_t *m_host_list = NULL; + m_host_list = calloc(2, sizeof(msg_host_t)); + + m_host_list[0] = MSG_host_self(); + m_host_list[1] = remote_host; + double computation_amount[] = { 0, 0 }; + double communication_amount[] = { 0, 0, 0, (double)size }; + + msg_task_t task = MSG_parallel_task_create("file transfer for write", 2, m_host_list, computation_amount, communication_amount, NULL); + msg_error_t transfer = MSG_parallel_task_execute(task); + MSG_task_destroy(task); + free(m_host_list); + if(transfer != MSG_OK){ + if (transfer == MSG_HOST_FAILURE) + XBT_WARN("Transfer error, remote host just turned off!"); + if (transfer == MSG_TASK_CANCELED) + XBT_WARN("Transfer error, task has been canceled!"); + + return -1; + } + } + /* Write file on local or remote host */ + write_size = simcall_file_write(file_priv->simdata->smx_file, size, remote_host); + + return write_size; } /** \ingroup msg_file_management @@ -126,7 +193,10 @@ msg_file_t MSG_file_open(const char* fullpath, void* data) priv->simdata = xbt_new0(s_simdata_file_t,1); priv->simdata->smx_file = simcall_file_open(fullpath); xbt_lib_set(file_lib, fullpath, MSG_FILE_LEVEL, priv); - return (msg_file_t) xbt_lib_get_elm_or_null(file_lib, fullpath); + msg_file_t fd = (msg_file_t) xbt_lib_get_elm_or_null(file_lib, fullpath); + __MSG_file_get_info(fd); + + return fd; } /** @@ -265,7 +335,72 @@ msg_error_t MSG_file_move (msg_file_t fd, const char* fullpath) msg_error_t MSG_file_rcopy (msg_file_t file, msg_host_t host, const char* fullpath) { msg_file_priv_t file_priv = MSG_file_priv(file); - return simcall_file_rcopy(file_priv->simdata->smx_file, host, fullpath); + + /* Find the host source where the file is physically located */ + msg_storage_t storage_src =(msg_storage_t) xbt_lib_get_elm_or_null(storage_lib, file_priv->storageId); + + msg_storage_priv_t storage_priv_src = MSG_storage_priv(storage_src); + const char *host_name_src, *host_name_dest; + host_name_src = storage_priv_src->host; + + /* Find the real host destination where the file will be physically stored */ + xbt_dict_cursor_t cursor = NULL; + char *mount_name, *storage_name, *file_mount_name; + msg_storage_t storage_dest; + size_t longest_prefix_length = 0; + + xbt_dict_t storage_list = simcall_host_get_mounted_storage_list(host); + xbt_dict_foreach(storage_list,cursor,mount_name,storage_name){ + file_mount_name = (char *) xbt_malloc ((strlen(mount_name)+1)); + strncpy(file_mount_name,fullpath,strlen(mount_name)+1); + file_mount_name[strlen(mount_name)] = '\0'; + + if(!strcmp(file_mount_name,mount_name) && strlen(mount_name)>longest_prefix_length){ + /* The current mount name is found in the full path and is bigger than the previous*/ + longest_prefix_length = strlen(mount_name); + storage_dest = (msg_storage_t) xbt_lib_get_elm_or_null(storage_lib, storage_name); + } + free(file_mount_name); + } + if(longest_prefix_length>0){ + /* Mount point found, retrieve the host the storage is attached to */ + msg_storage_priv_t storage_dest_priv = MSG_storage_priv(storage_dest); + host_name_dest = storage_dest_priv->host; + + }else{ + XBT_WARN("Can't find mount point for '%s' on destination host '%s'", fullpath, SIMIX_host_get_name(host)); + return MSG_TASK_CANCELED; + } + XBT_INFO("HOST SRC %s HOST DEST %s", host_name_src, host_name_dest); + + /* Try to send file calling SIMIX network layer */ + + size_t file_size = simcall_file_read(file_priv->simdata->smx_file, file_priv->size, MSG_get_host_by_name(host_name_src)); + xbt_ex_t e; + msg_error_t ret = MSG_OK; + + TRY { + msg_mailbox_t mailbox = MSG_mailbox_get_by_alias(host_name_dest); + simcall_comm_isend(mailbox, (double) file_size, -1.0, file, sizeof(void *), NULL, NULL, (void*)file, 0); + simcall_comm_irecv(mailbox, file, NULL, NULL, NULL, -1.0); + } + CATCH(e) { + switch (e.category) { + case cancel_error: + ret = MSG_HOST_FAILURE; + break; + case network_error: + ret = MSG_TRANSFER_FAILURE; + break; + case timeout_error: + ret = MSG_TIMEOUT; + break; + default: + RETHROW; + } + xbt_ex_free(e); + } + return ret; } /** diff --git a/src/simix/simcalls.in b/src/simix/simcalls.in index f22c4447a7..95711bd847 100644 --- a/src/simix/simcalls.in +++ b/src/simix/simcalls.in @@ -108,8 +108,8 @@ sem_would_block True (int) (sem, void*, smx_sem_t) sem_acquire False (void) (sem, void*, smx_sem_t) sem_acquire_timeout False (void) (sem, void*, smx_sem_t) (timeout, double) sem_get_capacity True (int) (sem, void*, smx_sem_t) -file_read False (sg_size_t) (fd, void*, smx_file_t) (size, sg_size_t) -file_write False (sg_size_t) (fd, void*, smx_file_t) (size, sg_size_t) +file_read False (sg_size_t) (fd, void*, smx_file_t) (size, sg_size_t) (host, void*, smx_host_t) +file_write False (sg_size_t) (fd, void*, smx_file_t) (size, sg_size_t) (host, void*, smx_host_t) file_open False (void*, smx_file_t) (fullpath, const char*) file_close False (int) (fd, void*, smx_file_t) file_unlink True (int) (fd, void*, smx_file_t) diff --git a/src/simix/simcalls_generated_args_getter_setter.h b/src/simix/simcalls_generated_args_getter_setter.h index 1e1c4219a0..b55da6338d 100644 --- a/src/simix/simcalls_generated_args_getter_setter.h +++ b/src/simix/simcalls_generated_args_getter_setter.h @@ -1064,6 +1064,12 @@ static inline sg_size_t simcall_file_read__get__size(smx_simcall_t simcall){ static inline void simcall_file_read__set__size(smx_simcall_t simcall, sg_size_t arg){ simcall->args[1].sgsz = arg; } +static inline smx_host_t simcall_file_read__get__host(smx_simcall_t simcall){ + return (smx_host_t) simcall->args[2].dp; +} +static inline void simcall_file_read__set__host(smx_simcall_t simcall, void* arg){ + simcall->args[2].dp = arg; +} static inline smx_file_t simcall_file_write__get__fd(smx_simcall_t simcall){ return (smx_file_t) simcall->args[0].dp; } @@ -1076,6 +1082,12 @@ static inline sg_size_t simcall_file_write__get__size(smx_simcall_t simcall){ static inline void simcall_file_write__set__size(smx_simcall_t simcall, sg_size_t arg){ simcall->args[1].sgsz = arg; } +static inline smx_host_t simcall_file_write__get__host(smx_simcall_t simcall){ + return (smx_host_t) simcall->args[2].dp; +} +static inline void simcall_file_write__set__host(smx_simcall_t simcall, void* arg){ + simcall->args[2].dp = arg; +} static inline const char* simcall_file_open__get__fullpath(smx_simcall_t simcall){ return simcall->args[0].cc; } diff --git a/src/simix/simcalls_generated_body.c b/src/simix/simcalls_generated_body.c index 885dc4b46f..f6f4fdbd2e 100644 --- a/src/simix/simcalls_generated_body.c +++ b/src/simix/simcalls_generated_body.c @@ -1610,13 +1610,14 @@ } return self->simcall.result.i; } - inline static sg_size_t simcall_BODY_file_read(smx_file_t fd, sg_size_t size) { + inline static sg_size_t simcall_BODY_file_read(smx_file_t fd, sg_size_t size, smx_host_t host) { smx_process_t self = SIMIX_process_self(); self->simcall.call = SIMCALL_FILE_READ; memset(&self->simcall.result, 0, sizeof(self->simcall.result)); memset(self->simcall.args, 0, sizeof(self->simcall.args)); self->simcall.args[0].dp = (void*) fd; self->simcall.args[1].sgsz = (sg_size_t) size; + self->simcall.args[2].dp = (void*) host; if (self != simix_global->maestro_process) { XBT_DEBUG("Yield process '%s' on simcall %s (%d)", self->name, SIMIX_simcall_name(self->simcall.call), (int)self->simcall.call); @@ -1626,13 +1627,14 @@ } return self->simcall.result.sgsz; } - inline static sg_size_t simcall_BODY_file_write(smx_file_t fd, sg_size_t size) { + inline static sg_size_t simcall_BODY_file_write(smx_file_t fd, sg_size_t size, smx_host_t host) { smx_process_t self = SIMIX_process_self(); self->simcall.call = SIMCALL_FILE_WRITE; memset(&self->simcall.result, 0, sizeof(self->simcall.result)); memset(self->simcall.args, 0, sizeof(self->simcall.args)); self->simcall.args[0].dp = (void*) fd; self->simcall.args[1].sgsz = (sg_size_t) size; + self->simcall.args[2].dp = (void*) host; if (self != simix_global->maestro_process) { XBT_DEBUG("Yield process '%s' on simcall %s (%d)", self->name, SIMIX_simcall_name(self->simcall.call), (int)self->simcall.call); diff --git a/src/simix/simcalls_generated_case.c b/src/simix/simcalls_generated_case.c index bf5a69959e..01ebce1c2e 100644 --- a/src/simix/simcalls_generated_case.c +++ b/src/simix/simcalls_generated_case.c @@ -502,11 +502,11 @@ case SIMCALL_SEM_GET_CAPACITY: break; case SIMCALL_FILE_READ: - SIMIX_pre_file_read(simcall , (smx_file_t) simcall->args[0].dp, simcall->args[1].sgsz); + SIMIX_pre_file_read(simcall , (smx_file_t) simcall->args[0].dp, simcall->args[1].sgsz, (smx_host_t) simcall->args[2].dp); break; case SIMCALL_FILE_WRITE: - SIMIX_pre_file_write(simcall , (smx_file_t) simcall->args[0].dp, simcall->args[1].sgsz); + SIMIX_pre_file_write(simcall , (smx_file_t) simcall->args[0].dp, simcall->args[1].sgsz, (smx_host_t) simcall->args[2].dp); break; case SIMCALL_FILE_OPEN: diff --git a/src/simix/smx_io.c b/src/simix/smx_io.c index b4dc9b8b6f..4be4d69fdb 100644 --- a/src/simix/smx_io.c +++ b/src/simix/smx_io.c @@ -50,17 +50,16 @@ void SIMIX_storage_destroy(void *s) } //SIMIX FILE READ -void SIMIX_pre_file_read(smx_simcall_t simcall, smx_file_t fd, sg_size_t size) +void SIMIX_pre_file_read(smx_simcall_t simcall, smx_file_t fd, sg_size_t size, smx_host_t host) { - smx_action_t action = SIMIX_file_read(simcall->issuer, fd, size); + smx_action_t action = SIMIX_file_read(simcall->issuer, fd, size, host); xbt_fifo_push(action->simcalls, simcall); simcall->issuer->waiting_action = action; } -smx_action_t SIMIX_file_read(smx_process_t process, smx_file_t fd, sg_size_t size) +smx_action_t SIMIX_file_read(smx_process_t process, smx_file_t fd, sg_size_t size, smx_host_t host) { smx_action_t action; - smx_host_t host = process->smx_host; /* check if the host is active */ if (surf_resource_get_state(surf_workstation_resource_priv(host)) != SURF_RESOURCE_ON) { @@ -85,17 +84,16 @@ smx_action_t SIMIX_file_read(smx_process_t process, smx_file_t fd, sg_size_t siz } //SIMIX FILE WRITE -void SIMIX_pre_file_write(smx_simcall_t simcall, smx_file_t fd, sg_size_t size) +void SIMIX_pre_file_write(smx_simcall_t simcall, smx_file_t fd, sg_size_t size, smx_host_t host) { - smx_action_t action = SIMIX_file_write(simcall->issuer, fd, size); + smx_action_t action = SIMIX_file_write(simcall->issuer, fd, size, host); xbt_fifo_push(action->simcalls, simcall); simcall->issuer->waiting_action = action; } -smx_action_t SIMIX_file_write(smx_process_t process, smx_file_t fd, sg_size_t size) +smx_action_t SIMIX_file_write(smx_process_t process, smx_file_t fd, sg_size_t size, smx_host_t host) { smx_action_t action; - smx_host_t host = process->smx_host; /* check if the host is active */ if (surf_resource_get_state(surf_workstation_resource_priv(host)) != SURF_RESOURCE_ON) { @@ -114,7 +112,7 @@ smx_action_t SIMIX_file_write(smx_process_t process, smx_file_t fd, sg_size_t si action->io.surf_io = surf_workstation_write(host, fd->surf_file, size); surf_action_set_data(action->io.surf_io, action); - XBT_DEBUG("Create io action %p", action); + XBT_INFO("Create io action %p", action); return action; } diff --git a/src/simix/smx_io_private.h b/src/simix/smx_io_private.h index 15e2624341..4ab6974a67 100644 --- a/src/simix/smx_io_private.h +++ b/src/simix/smx_io_private.h @@ -22,8 +22,8 @@ static inline smx_storage_priv_t SIMIX_storage_priv(smx_storage_t storage){ smx_storage_t SIMIX_storage_create(const char *name, void *storage, void *data); void SIMIX_storage_destroy(void *s); -void SIMIX_pre_file_read(smx_simcall_t simcall, smx_file_t fd, sg_size_t size); -void SIMIX_pre_file_write(smx_simcall_t simcall,smx_file_t fd, sg_size_t size); +void SIMIX_pre_file_read(smx_simcall_t simcall, smx_file_t fd, sg_size_t size, smx_host_t host); +void SIMIX_pre_file_write(smx_simcall_t simcall,smx_file_t fd, sg_size_t size, smx_host_t host); void SIMIX_pre_file_open(smx_simcall_t simcall, const char* fullpath); void SIMIX_pre_file_close(smx_simcall_t simcall, smx_file_t fd); int SIMIX_pre_file_unlink(smx_simcall_t simcall, smx_file_t fd); @@ -35,8 +35,8 @@ xbt_dynar_t SIMIX_pre_file_get_info(smx_simcall_t simcall, smx_file_t fd); int SIMIX_pre_file_seek(smx_simcall_t simcall, smx_file_t fd, sg_size_t offset, int origin); int SIMIX_pre_file_move(smx_simcall_t simcall, smx_file_t fd, const char* fullpath); int SIMIX_pre_file_rcopy(smx_simcall_t simcall, smx_file_t fd, smx_host_t dest, const char* fullpath); -smx_action_t SIMIX_file_read(smx_process_t process, smx_file_t fd, sg_size_t size); -smx_action_t SIMIX_file_write(smx_process_t process, smx_file_t fd, sg_size_t size); +smx_action_t SIMIX_file_read(smx_process_t process, smx_file_t fd, sg_size_t size, smx_host_t host); +smx_action_t SIMIX_file_write(smx_process_t process, smx_file_t fd, sg_size_t size, smx_host_t host); smx_action_t SIMIX_file_open(smx_process_t process, const char* fullpath); smx_action_t SIMIX_file_close(smx_process_t process, smx_file_t fd); int SIMIX_file_unlink(smx_process_t process, smx_file_t fd); diff --git a/src/simix/smx_user.c b/src/simix/smx_user.c index 43085e4746..c6da787256 100644 --- a/src/simix/smx_user.c +++ b/src/simix/smx_user.c @@ -1318,18 +1318,18 @@ int simcall_sem_get_capacity(smx_sem_t sem) * \ingroup simix_file_management * */ -sg_size_t simcall_file_read(smx_file_t fd, sg_size_t size) +sg_size_t simcall_file_read(smx_file_t fd, sg_size_t size, smx_host_t host) { - return simcall_BODY_file_read(fd, size); + return simcall_BODY_file_read(fd, size, host); } /** * \ingroup simix_file_management * */ -sg_size_t simcall_file_write(smx_file_t fd, sg_size_t size) +sg_size_t simcall_file_write(smx_file_t fd, sg_size_t size, smx_host_t host) { - return simcall_BODY_file_write(fd, size); + return simcall_BODY_file_write(fd, size, host); } /**