From 8d0cddd21e7461a4bc26cac0c9d391dcfcf9a891 Mon Sep 17 00:00:00 2001 From: Mohamad Chaarawi Date: Wed, 13 Jun 2018 18:37:37 +0000 Subject: [PATCH] Add DFS ior/mdtest driver Signed-off-by: Mohamad Chaarawi --- configure.ac | 19 ++ src/Makefile.am | 4 + src/aiori-DFS.c | 562 ++++++++++++++++++++++++++++++++++++++++++++++++ src/aiori.c | 3 + src/aiori.h | 4 + src/ior.c | 11 +- src/mdtest.c | 10 + 7 files changed, 612 insertions(+), 1 deletion(-) create mode 100755 src/aiori-DFS.c diff --git a/configure.ac b/configure.ac index 4f9461c..af6e49a 100755 --- a/configure.ac +++ b/configure.ac @@ -142,8 +142,27 @@ AM_COND_IF([USE_POSIX_AIORI],[ AC_DEFINE([USE_POSIX_AIORI], [], [Build POSIX backend AIORI]) ]) +# DFS IO support +AC_ARG_WITH([daos], + [AS_HELP_STRING([--with-daos], + [support IO with DFS backend @<:@default=no@:>@])], + [], + [with_daos=no]) +AS_IF([test "x$with_daos" != xno], + DAOS="yes" + LDFLAGS="$LDFLAGS -L$with_daos/lib" + CPPFLAGS="$CPPFLAGS -I$with_daos/include" + AC_CHECK_HEADERS(daos_types.h,, [unset DAOS]) + AC_CHECK_LIB([uuid], [uuid_generate],, [unset DAOS]) + AC_CHECK_LIB([daos_common], [daos_sgl_init],, [unset DAOS]) + AC_CHECK_LIB([daos], [daos_init],, [unset DAOS]) + AC_CHECK_LIB([dfs], [dfs_mkdir],, [unset DAOS])) +AM_CONDITIONAL([USE_DFS_AIORI], [test x$DAOS = xyes]) +AM_COND_IF([USE_DFS_AIORI],[ + AC_DEFINE([USE_DFS_AIORI], [], [Build DFS backend AIORI]) +]) # aws4c is needed for the S3 backend (see --with-S3, below). # Version 0.5.2 of aws4c is available at https://github.com/jti-lanl/aws4c.git diff --git a/src/Makefile.am b/src/Makefile.am index 7d2575b..aea1824 100755 --- a/src/Makefile.am +++ b/src/Makefile.am @@ -53,6 +53,10 @@ if USE_POSIX_AIORI extraSOURCES += aiori-POSIX.c endif +if USE_DFS_AIORI +extraSOURCES += aiori-DFS.c +endif + if USE_S3_AIORI extraSOURCES += aiori-S3.c diff --git a/src/aiori-DFS.c b/src/aiori-DFS.c new file mode 100755 index 0000000..ad0c712 --- /dev/null +++ b/src/aiori-DFS.c @@ -0,0 +1,562 @@ +/* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*- + * vim:expandtab:shiftwidth=8:tabstop=8: + */ +/******************************************************************************\ +* * +* Copyright (c) 2003, The Regents of the University of California * +* See the file COPYRIGHT for a complete copyright notice and license. * +* * +******************************************************************************** +* +* Implement of abstract I/O interface for DFS. +* +\******************************************************************************/ + +#ifdef HAVE_CONFIG_H +#include "config.h" +#endif + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include + +#include "ior.h" +#include "aiori.h" +#include "iordef.h" +#include "utilities.h" + +dfs_t *dfs; + +static int +parse_filename(const char *path, char **_obj_name, char **_cont_name) +{ + char *f1 = NULL; + char *f2 = NULL; + char *fname = NULL; + char *cont_name = NULL; + int rc = 0; + + if (path == NULL || _obj_name == NULL || _cont_name == NULL) + return -EINVAL; + + if (strcmp(path, "/") == 0) { + *_cont_name = strdup("/"); + if (*_cont_name == NULL) + return -ENOMEM; + *_obj_name = NULL; + return 0; + } + + f1 = strdup(path); + if (f1 == NULL) + D_GOTO(out, rc = -ENOMEM); + + f2 = strdup(path); + if (f2 == NULL) + D_GOTO(out, rc = -ENOMEM); + + fname = basename(f1); + cont_name = dirname(f2); + + if (cont_name[0] == '.' || cont_name[0] != '/') { + char *cwd; + + //getcwd(cwd, 1024); + cwd = strdup("/"); + if (strcmp(cont_name, ".") == 0) { + cont_name = strdup(cwd); + if (cont_name == NULL) + D_GOTO(out, rc = -ENOMEM); + } else { + char *new_dir = calloc(strlen(cwd) + strlen(cont_name) + + 1, sizeof(char)); + if (new_dir == NULL) + D_GOTO(out, rc = -ENOMEM); + + strcpy(new_dir, cwd); + if (cont_name[0] == '.') { + strcat(new_dir, &cont_name[1]); + } else { + strcat(new_dir, "/"); + strcat(new_dir, cont_name); + } + cont_name = new_dir; + } + *_cont_name = cont_name; + } else { + *_cont_name = strdup(cont_name); + if (*_cont_name == NULL) + D_GOTO(out, rc = -ENOMEM); + } + + *_obj_name = strdup(fname); + if (*_obj_name == NULL) { + free(*_cont_name); + *_cont_name = NULL; + D_GOTO(out, rc = -ENOMEM); + } + +out: + if (f1) + free(f1); + if (f2) + free(f2); + return rc; +} + +/**************************** P R O T O T Y P E S *****************************/ +static void *DFS_Create(char *, IOR_param_t *); +static void *DFS_Open(char *, IOR_param_t *); +static IOR_offset_t DFS_Xfer(int, void *, IOR_size_t *, + IOR_offset_t, IOR_param_t *); +static void DFS_Close(void *, IOR_param_t *); +static void DFS_Delete(char *, IOR_param_t *); +static void DFS_SetVersion(IOR_param_t *); +static void DFS_Fsync(void *, IOR_param_t *); +static IOR_offset_t DFS_GetFileSize(IOR_param_t *, MPI_Comm, char *); +static int DFS_Statfs (const char *, ior_aiori_statfs_t *, IOR_param_t *); +static int DFS_Stat (const char *, struct stat *, IOR_param_t *); +static int DFS_Mkdir (const char *, mode_t, IOR_param_t *); +static int DFS_Rmdir (const char *, IOR_param_t *); +static int DFS_Access (const char *, int, IOR_param_t *); + +/************************** D E C L A R A T I O N S ***************************/ + +ior_aiori_t dfs_aiori = { + .name = "DFS", + .create = DFS_Create, + .open = DFS_Open, + .xfer = DFS_Xfer, + .close = DFS_Close, + .delete = DFS_Delete, + .set_version = DFS_SetVersion, + .fsync = DFS_Fsync, + .get_file_size = DFS_GetFileSize, + .statfs = DFS_Statfs, + .mkdir = DFS_Mkdir, + .rmdir = DFS_Rmdir, + .access = DFS_Access, + .stat = DFS_Stat, +}; + +/***************************** F U N C T I O N S ******************************/ + +int +dfs_init(void) { + int rc; + + rc = daos_init(); + if (rc) { + fprintf(stderr, "daos_init() failed with %d\n", rc); + return rc; + } + + rc = dfs_mount(&dfs); + if (rc) { + fprintf(stderr, "dfs_mount failed (%d)\n", rc); + return 1; + } + + return rc; +} + +int dfs_finalize(void) +{ + dfs_umount(dfs); + daos_fini(); + return 0; +} + +/* + * Creat and open a file through the DFS interface. + */ +static void * +DFS_Create(char *testFileName, IOR_param_t *param) +{ + char *name = NULL, *dir_name = NULL; + dfs_obj_t *obj = NULL, *parent = NULL; + mode_t pmode; + int fd_oflag = 0; + int rc; + + fd_oflag |= O_CREAT | O_RDWR; + + rc = parse_filename(testFileName, &name, &dir_name); + if (rc) + goto out; + + assert(dir_name); + assert(name); + + rc = dfs_lookup(dfs, dir_name, &parent, &pmode); + if (rc || !S_ISDIR(pmode)) + goto out; + + mode_t mode = S_IFREG | param->mode; + rc = dfs_open(dfs, parent, name, mode, fd_oflag, NULL, &obj); + if (rc) + goto out; + +out: + if (name) + free(name); + if (dir_name) + free(dir_name); + if (parent) + dfs_release(parent); + + return ((void *)obj); +} + +/* + * Open a file through the DFS interface. + */ +static void *DFS_Open(char *testFileName, IOR_param_t *param) +{ + char *name = NULL, *dir_name = NULL; + dfs_obj_t *obj = NULL, *parent = NULL; + mode_t pmode; + int rc; + int fd_oflag = 0; + + fd_oflag |= O_RDWR; + + rc = parse_filename(testFileName, &name, &dir_name); + if (rc) + goto out; + + assert(dir_name); + assert(name); + + rc = dfs_lookup(dfs, dir_name, &parent, &pmode); + if (rc || !S_ISDIR(pmode)) + goto out; + + rc = dfs_open(dfs, parent, name, S_IFREG, fd_oflag, NULL, &obj); + if (rc) + goto out; + +out: + if (name) + free(name); + if (dir_name) + free(dir_name); + if (parent) + dfs_release(parent); + + return ((void *)obj); +} + +/* + * Write or read access to file using the DFS interface. + */ +static IOR_offset_t +DFS_Xfer(int access, void *file, IOR_size_t *buffer, IOR_offset_t length, + IOR_param_t *param) +{ + int xferRetries = 0; + long long remaining = (long long)length; + char *ptr = (char *)buffer; + daos_size_t ret; + int rc; + dfs_obj_t *obj; + + obj = (dfs_obj_t *)file; + + while (remaining > 0) { + daos_iov_t iov; + daos_sg_list_t sgl; + + /** set memory location */ + sgl.sg_nr = 1; + sgl.sg_nr_out = 0; + daos_iov_set(&iov, (void *)ptr, remaining); + sgl.sg_iovs = &iov; + + /* write/read file */ + if (access == WRITE) { + rc = dfs_write(dfs, obj, sgl, param->offset); + if (rc) + ERR("write() failed"); + ret = remaining; + } else { + rc = dfs_read(dfs, obj, sgl, param->offset, &ret); + if (rc || ret == 0) + ERR("read() failed"); + } + + if (ret < remaining) { + if (param->singleXferAttempt == TRUE) + MPI_CHECK(MPI_Abort(MPI_COMM_WORLD, -1), + "barrier error"); + if (xferRetries > MAX_RETRY) + ERR("too many retries -- aborting"); + } + + assert(ret >= 0); + assert(ret <= remaining); + remaining -= ret; + ptr += ret; + xferRetries++; + } + + return (length); +} + +/* + * Perform fsync(). + */ +static void DFS_Fsync(void *fd, IOR_param_t * param) +{ + return; +} + +/* + * Close a file through the DFS interface. + */ +static void DFS_Close(void *fd, IOR_param_t * param) +{ + dfs_release((dfs_obj_t *)fd); +} + +/* + * Delete a file through the DFS interface. + */ +static void DFS_Delete(char *testFileName, IOR_param_t * param) +{ + char *name = NULL, *dir_name = NULL; + dfs_obj_t *parent = NULL; + mode_t pmode; + int rc; + + rc = parse_filename(testFileName, &name, &dir_name); + if (rc) + goto out; + + assert(dir_name); + assert(name); + + rc = dfs_lookup(dfs, dir_name, &parent, &pmode); + if (rc || !S_ISDIR(pmode)) + goto out; + + rc = dfs_remove(dfs, parent, name); + if (rc) + goto out; + +out: + if (name) + free(name); + if (dir_name) + free(dir_name); + if (parent) + dfs_release(parent); +} + +/* + * Determine api version. + */ +static void DFS_SetVersion(IOR_param_t * test) +{ + strcpy(test->apiVersion, test->api); +} + +/* + * Use DFS stat() to return aggregate file size. + */ +static IOR_offset_t DFS_GetFileSize(IOR_param_t * test, MPI_Comm testComm, + char *testFileName) +{ + dfs_obj_t *obj; + daos_size_t fsize, tmpMin, tmpMax, tmpSum; + int rc; + + rc = dfs_lookup(dfs, testFileName, &obj, NULL); + if (rc) + return -1; + + rc = dfs_get_size(dfs, obj, &fsize); + if (rc) + return -1; + + dfs_release(obj); + + if (test->filePerProc == TRUE) { + MPI_CHECK(MPI_Allreduce(&fsize, &tmpSum, 1, + MPI_LONG_LONG_INT, MPI_SUM, testComm), + "cannot total data moved"); + fsize = tmpSum; + } else { + MPI_CHECK(MPI_Allreduce(&fsize, &tmpMin, 1, + MPI_LONG_LONG_INT, MPI_MIN, testComm), + "cannot total data moved"); + MPI_CHECK(MPI_Allreduce(&fsize, &tmpMax, 1, + MPI_LONG_LONG_INT, MPI_MAX, testComm), + "cannot total data moved"); + if (tmpMin != tmpMax) { + if (rank == 0) { + WARN("inconsistent file size by different tasks"); + } + /* incorrect, but now consistent across tasks */ + fsize = tmpMin; + } + } + + return (fsize); +} + +static int +DFS_Statfs(const char *path, ior_aiori_statfs_t *sfs, IOR_param_t * param) +{ + return 0; +} + +static int +DFS_Mkdir (const char *path, mode_t mode, IOR_param_t * param) +{ + dfs_obj_t *parent = NULL; + mode_t pmode; + char *name = NULL, *dir_name = NULL; + int rc; + + rc = parse_filename(path, &name, &dir_name); + if (rc) + return rc; + + assert(dir_name); + assert(name); + + rc = dfs_lookup(dfs, dir_name, &parent, &pmode); + if (rc || !S_ISDIR(pmode)) + goto out; + + rc = dfs_mkdir(dfs, parent, name, mode); + if (rc) + goto out; + +out: + if (name) + free(name); + if (dir_name) + free(dir_name); + if (parent) + dfs_release(parent); + return rc; +} + +static int +DFS_Rmdir (const char *path, IOR_param_t * param) +{ + dfs_obj_t *parent = NULL; + mode_t pmode; + char *name = NULL, *dir_name = NULL; + int rc; + + rc = parse_filename(path, &name, &dir_name); + if (rc) + return rc; + + assert(dir_name); + assert(name); + + rc = dfs_lookup(dfs, dir_name, &parent, &pmode); + if (rc || !S_ISDIR(pmode)) + goto out; + + rc = dfs_remove(dfs, parent, name); + if (rc) + goto out; + +out: + if (name) + free(name); + if (dir_name) + free(dir_name); + if (parent) + dfs_release(parent); + return rc; +} + +static int +DFS_Access (const char *path, int mode, IOR_param_t * param) +{ + dfs_obj_t *parent = NULL; + mode_t pmode; + char *name = NULL, *dir_name = NULL; + struct stat stbuf; + int rc; + + rc = parse_filename(path, &name, &dir_name); + if (rc) + return rc; + + assert(dir_name); + assert(name); + + rc = dfs_lookup(dfs, dir_name, &parent, &pmode); + if (rc || !S_ISDIR(pmode)) + goto out; + + if (strcmp(name, ".") == 0) { + free(name); + name = NULL; + } + rc = dfs_stat(dfs, parent, name, &stbuf); + if (rc) { + rc = -1; + errno = -ENOENT; + goto out; + } + +out: + if (name) + free(name); + if (dir_name) + free(dir_name); + if (parent) + dfs_release(parent); + return rc; +} + +static int +DFS_Stat (const char *path, struct stat *buf, IOR_param_t * param) +{ + dfs_obj_t *parent = NULL; + mode_t pmode; + char *name = NULL, *dir_name = NULL; + int rc; + + rc = parse_filename(path, &name, &dir_name); + if (rc) + return rc; + + assert(dir_name); + assert(name); + + rc = dfs_lookup(dfs, dir_name, &parent, &pmode); + if (rc || !S_ISDIR(pmode)) + goto out; + + rc = dfs_stat(dfs, parent, name, buf); + if (rc) + goto out; + +out: + if (name) + free(name); + if (dir_name) + free(dir_name); + if (parent) + dfs_release(parent); + return rc; +} diff --git a/src/aiori.c b/src/aiori.c index 677c1ea..a40cbd6 100644 --- a/src/aiori.c +++ b/src/aiori.c @@ -51,6 +51,9 @@ ior_aiori_t *available_aiori[] = { &s3_aiori, &s3_plus_aiori, &s3_emc_aiori, +#endif +#ifdef USE_DFS_AIORI + &dfs_aiori, #endif NULL }; diff --git a/src/aiori.h b/src/aiori.h index 4ee400a..9e5695e 100755 --- a/src/aiori.h +++ b/src/aiori.h @@ -88,11 +88,15 @@ extern ior_aiori_t mmap_aiori; extern ior_aiori_t s3_aiori; extern ior_aiori_t s3_plus_aiori; extern ior_aiori_t s3_emc_aiori; +extern ior_aiori_t dfs_aiori; const ior_aiori_t *aiori_select (const char *api); int aiori_count (void); const char *aiori_default (void); +int dfs_init(void); +int dfs_finalize(void); + IOR_offset_t MPIIO_GetFileSize(IOR_param_t * test, MPI_Comm testComm, char *testFileName); diff --git a/src/ior.c b/src/ior.c index b92b40d..e4dc03b 100755 --- a/src/ior.c +++ b/src/ior.c @@ -733,7 +733,7 @@ static void DisplayUsage(char **argv) { char *opts[] = { "OPTIONS:", - " -a S api -- API for I/O [POSIX|MMAP|MPIIO|HDF5|HDFS|S3|S3_EMC|NCMPI]", + " -a S api -- API for I/O [POSIX|DFS|MMAP|MPIIO|HDF5|HDFS|S3|S3_EMC|NCMPI]", " -A N refNum -- user supplied reference number to include in the summary", " -b N blockSize -- contiguous bytes to write per task (e.g.: 8, 4k, 2m, 1g)", " -B useO_DIRECT -- uses O_DIRECT for POSIX, bypassing I/O buffers", @@ -2037,6 +2037,11 @@ static void TestIoSys(IOR_test_t *test) /* bind I/O calls to specific API */ AioriBind(params->api, params); +#ifdef USE_DFS_AIORI + if (strcmp(params->api, "DFS") == 0) + dfs_init(); +#endif + /* show test setup */ if (rank == 0 && verbose >= VERBOSE_0) ShowSetup(params); @@ -2310,6 +2315,10 @@ static void TestIoSys(IOR_test_t *test) /* Sync with the tasks that did not participate in this test */ MPI_CHECK(MPI_Barrier(MPI_COMM_WORLD), "barrier error"); +#ifdef USE_DFS_AIORI + if (strcmp(params->api, "DFS") == 0) + dfs_finalize(); +#endif } /* diff --git a/src/mdtest.c b/src/mdtest.c index aadbf78..3245c47 100644 --- a/src/mdtest.c +++ b/src/mdtest.c @@ -1936,6 +1936,11 @@ int main(int argc, char **argv) { } } +#ifdef USE_DFS_AIORI + if (strcmp(backend_name, "DFS") == 0) + dfs_init(); +#endif + if (!create_only && !stat_only && !read_only && !remove_only) { create_only = stat_only = read_only = remove_only = 1; if (( rank == 0 ) && ( verbose >= 1 )) { @@ -2411,6 +2416,11 @@ int main(int argc, char **argv) { free(rand_array); } +#ifdef USE_DFS_AIORI + if (strcmp(backend_name, "DFS") == 0) + dfs_finalize(); +#endif + MPI_Finalize(); exit(0); }