vitalif
/
mdtest
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Merge pull request #1 from mchaarawi/master 

add DFS plugin and update DAOS plugin
master
Vishwanath Venkatesan 2018-08-29 15:32:59 -07:00 committed by GitHub
parent 98ae5df16d ced745ae4f
commit 0adc246adc
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
14 changed files with 851 additions and 137 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
.gitignore vendored
View File

@ -4,6 +4,8 @@ Makefile.in
aclocal.m4
config.log
config.status
COPYING
INSTALL
config/compile
config/config.guess
config/config.sub
@ -13,12 +15,14 @@ config/missing
config/test-driver
configure
contrib/.deps/
contrib/cbif
contrib/Makefile
contrib/Makefile.in
contrib/cbif
doc/Makefile
doc/Makefile.in
src/.deps/
src/mdtest
src/Makefile
src/Makefile.in
src/config.h

28
configure.ac
View File

@ -166,17 +166,27 @@ AM_COND_IF([USE_RADOS_AIORI],[
AC_DEFINE([USE_RADOS_AIORI], [], [Build RADOS backend AIORI])
])
# DAOS support
# DAOS Backends (DAOS and DFS) IO support
AC_ARG_WITH([daos],
[AS_HELP_STRING([--with-daos],
[support IO with DAOS backend @<:@default=no@:>@])],
[],
[with_daos=no])
AM_CONDITIONAL([USE_DAOS_AIORI], [test x$with_daos = xyes])
AM_COND_IF([USE_DAOS_AIORI],[
AC_DEFINE([USE_DAOS_AIORI], [], [Build DAOS backend AIORI])
])
[AS_HELP_STRING([--with-daos],
[support IO with DAOS backends @<:@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_DAOS_AIORI], [test x$DAOS = xyes])
AM_COND_IF([USE_DAOS_AIORI],[
AC_DEFINE([USE_DAOS_AIORI], [], [Build DAOS backends 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

3
src/Makefile.am
View File

@ -71,8 +71,7 @@ extraLDADD += -lrados
endif
if USE_DAOS_AIORI
extraSOURCES += aiori-DAOS.c list.h
extraLDADD += -ldaos -ldaos_common -luuid
extraSOURCES += aiori-DAOS.c aiori-DFS.c list.h
endif
if USE_S3_AIORI

191
src/aiori-DAOS.c
View File

@ -35,6 +35,56 @@
#include "iordef.h"
#include "list.h"
/************************** O P T I O N S *****************************/
struct daos_options{
char *daosPool;
char *daosPoolSvc;
char *daosGroup;
int daosRecordSize;
int daosStripeSize;
uint64_t daosStripeCount;
uint64_t daosStripeMax; /* max length of a stripe */
int daosAios; /* max number of concurrent async I/Os */
int daosWriteOnly; /* write only, no flush and commit */
uint64_t daosEpoch; /* epoch to access */
uint64_t daosWait; /* epoch to wait for before reading */
int daosKill; /* kill a target while running IOR */
char *daosObjectClass; /* object class */
};
static struct daos_options o = {
.daosPool = NULL,
.daosPoolSvc = NULL,
.daosGroup = NULL,
.daosRecordSize = 262144,
.daosStripeSize = 524288,
.daosStripeCount = -1,
.daosStripeMax = 0,
.daosAios = 1,
.daosWriteOnly = 0,
.daosEpoch = 0,
.daosWait = 0,
.daosKill = 0,
.daosObjectClass = NULL,
};
static option_help options [] = {
{'p', "daosPool", "pool uuid", OPTION_REQUIRED_ARGUMENT, 's', &o.daosPool},
{'v', "daosPoolSvc", "pool SVCL", OPTION_REQUIRED_ARGUMENT, 's', &o.daosPoolSvc},
{'g', "daosGroup", "server group", OPTION_OPTIONAL_ARGUMENT, 's', &o.daosGroup},
{'r', "daosRecordSize", "Record Size", OPTION_OPTIONAL_ARGUMENT, 'd', &o.daosRecordSize},
{'s', "daosStripeSize", "Stripe Size", OPTION_OPTIONAL_ARGUMENT, 'd', &o.daosStripeSize},
{'c', "daosStripeCount", "Stripe Count", OPTION_OPTIONAL_ARGUMENT, 'u', &o.daosStripeCount},
{'m', "daosStripeMax", "Max Stripe",OPTION_OPTIONAL_ARGUMENT, 'u', &o.daosStripeMax},
{'a', "daosAios", "Concurrent Async IOs",OPTION_OPTIONAL_ARGUMENT, 'd', &o.daosAios},
{'w', "daosWriteOnly", "Write Only, no commit",OPTION_OPTIONAL_ARGUMENT, 'd', &o.daosWriteOnly},
{'e', "daosEpoch", "Epoch Number to Access",OPTION_OPTIONAL_ARGUMENT, 'u', &o.daosEpoch},
{'t', "daosWait", "Epoch to wait for before read",OPTION_OPTIONAL_ARGUMENT, 'u', &o.daosWait},
{'k', "daosKill", "Kill target while running",OPTION_OPTIONAL_ARGUMENT, 'd', &o.daosKill},
{'o', "daosObjectClass", "object class", OPTION_OPTIONAL_ARGUMENT, 's', &o.daosObjectClass},
LAST_OPTION
};
/**************************** P R O T O T Y P E S *****************************/
static void DAOS_Init(IOR_param_t *);
@ -48,21 +98,23 @@ static void DAOS_Delete(char *, IOR_param_t *);
static char* DAOS_GetVersion();
static void DAOS_Fsync(void *, IOR_param_t *);
static IOR_offset_t DAOS_GetFileSize(IOR_param_t *, MPI_Comm, char *);
static option_help * DAOS_options();
/************************** D E C L A R A T I O N S ***************************/
ior_aiori_t daos_aiori = {
.name = "DAOS",
.create = DAOS_Create,
.open = DAOS_Open,
.xfer = DAOS_Xfer,
.close = DAOS_Close,
.delete = DAOS_Delete,
.get_version = DAOS_GetVersion,
.fsync = DAOS_Fsync,
.get_file_size = DAOS_GetFileSize,
.initialize = DAOS_Init,
.finalize = DAOS_Fini,
.name = "DAOS",
.create = DAOS_Create,
.open = DAOS_Open,
.xfer = DAOS_Xfer,
.close = DAOS_Close,
.delete = DAOS_Delete,
.get_version = DAOS_GetVersion,
.fsync = DAOS_Fsync,
.get_file_size = DAOS_GetFileSize,
.initialize = DAOS_Init,
.finalize = DAOS_Fini,
.get_options = DAOS_options,
};
enum handleType {
@ -234,10 +286,10 @@ static void ContainerOpen(char *testFileName, IOR_param_t *param,
info->ci_epoch_state.es_ghpce);
#if 0
if (param->open != WRITE && param->daosWait != 0) {
if (param->open != WRITE && o.daosWait != 0) {
daos_epoch_t e;
e = param->daosWait;
e = o.daosWait;
INFO(VERBOSE_2, param, "Waiting for epoch %lu", e);
@ -245,7 +297,7 @@ static void ContainerOpen(char *testFileName, IOR_param_t *param,
NULL /* ignore HLE */,
NULL /* synchronous */);
DCHECK(rc, "Failed to wait for epoch %lu",
param->daosWait);
o.daosWait);
}
if (param->open == WRITE &&
@ -345,10 +397,10 @@ static void AIOInit(IOR_param_t *param)
int rc;
rc = posix_memalign((void **) &buffers, sysconf(_SC_PAGESIZE),
param->transferSize * param->daosAios);
param->transferSize * o.daosAios);
DCHECK(rc, "Failed to allocate buffer array");
for (i = 0; i < param->daosAios; i++) {
for (i = 0; i < o.daosAios; i++) {
aio = malloc(sizeof *aio);
if (aio == NULL)
ERR("Failed to allocate aio array");
@ -394,9 +446,9 @@ static void AIOInit(IOR_param_t *param)
aio->a_iov.iov_buf);
}
nAios = param->daosAios;
nAios = o.daosAios;
events = malloc((sizeof *events) * param->daosAios);
events = malloc((sizeof *events) * o.daosAios);
if (events == NULL)
ERR("Failed to allocate events array");
}
@ -425,10 +477,10 @@ static void AIOWait(IOR_param_t *param)
int i;
int rc;
rc = daos_eq_poll(eventQueue, 0, DAOS_EQ_WAIT, param->daosAios,
rc = daos_eq_poll(eventQueue, 0, DAOS_EQ_WAIT, o.daosAios,
events);
DCHECK(rc, "Failed to poll event queue");
assert(rc <= param->daosAios - nAios);
assert(rc <= o.daosAios - nAios);
for (i = 0; i < rc; i++) {
int ret;
@ -455,7 +507,7 @@ static void AIOWait(IOR_param_t *param)
}
INFO(VERBOSE_3, param, "Found %d completed AIOs (%d free %d busy)", rc,
nAios, param->daosAios - nAios);
nAios, o.daosAios - nAios);
}
static void ObjectClassParse(const char *string)
@ -486,18 +538,11 @@ static void ObjectClassParse(const char *string)
GERR("Invalid 'daosObjectClass' argument: '%s'", string);
}
static const char *GetGroup(IOR_param_t *param)
{
if (strlen(param->daosGroup) == 0)
return NULL;
return param->daosGroup;
}
static void ParseService(IOR_param_t *param, int max, d_rank_list_t *ranks)
{
char *s;
s = strdup(param->daosPoolSvc);
s = strdup(o.daosPoolSvc);
if (s == NULL)
GERR("failed to duplicate argument");
ranks->rl_nr = 0;
@ -513,22 +558,26 @@ static void ParseService(IOR_param_t *param, int max, d_rank_list_t *ranks)
free(s);
}
static option_help * DAOS_options(){
return options;
}
static void DAOS_Init(IOR_param_t *param)
{
int rc;
if (strlen(param->daosObjectClass) != 0)
ObjectClassParse(param->daosObjectClass);
if (o.daosObjectClass)
ObjectClassParse(o.daosObjectClass);
if (param->filePerProc)
GERR("'filePerProc' not yet supported");
if (param->daosStripeMax % param->daosStripeSize != 0)
if (o.daosStripeMax % o.daosStripeSize != 0)
GERR("'daosStripeMax' must be a multiple of 'daosStripeSize'");
if (param->daosStripeSize % param->transferSize != 0)
if (o.daosStripeSize % param->transferSize != 0)
GERR("'daosStripeSize' must be a multiple of 'transferSize'");
if (param->transferSize % param->daosRecordSize != 0)
if (param->transferSize % o.daosRecordSize != 0)
GERR("'transferSize' must be a multiple of 'daosRecordSize'");
if (param->daosKill && ((objectClass != DAOS_OC_R2_RW) ||
if (o.daosKill && ((objectClass != DAOS_OC_R2_RW) ||
(objectClass != DAOS_OC_R3_RW) ||
(objectClass != DAOS_OC_R4_RW) ||
(objectClass != DAOS_OC_R2S_RW) ||
@ -551,23 +600,23 @@ static void DAOS_Init(IOR_param_t *param)
d_rank_t rank[13];
d_rank_list_t ranks;
if (strlen(param->daosPool) == 0)
if (o.daosPool == NULL)
GERR("'daosPool' must be specified");
if (strlen(param->daosPoolSvc) == 0)
if (o.daosPoolSvc == NULL)
GERR("'daosPoolSvc' must be specified");
INFO(VERBOSE_2, param, "Connecting to pool %s %s",
param->daosPool, param->daosPoolSvc);
o.daosPool, o.daosPoolSvc);
rc = uuid_parse(param->daosPool, uuid);
DCHECK(rc, "Failed to parse 'daosPool': %s", param->daosPool);
rc = uuid_parse(o.daosPool, uuid);
DCHECK(rc, "Failed to parse 'daosPool': %s", o.daosPool);
ranks.rl_ranks = rank;
ParseService(param, sizeof(rank) / sizeof(rank[0]), &ranks);
rc = daos_pool_connect(uuid, GetGroup(param), &ranks,
rc = daos_pool_connect(uuid, o.daosGroup, &ranks,
DAOS_PC_RW, &pool, &poolInfo,
NULL /* ev */);
DCHECK(rc, "Failed to connect to pool %s", param->daosPool);
DCHECK(rc, "Failed to connect to pool %s", o.daosPool);
}
HandleDistribute(&pool, POOL_HANDLE, param);
@ -576,8 +625,8 @@ static void DAOS_Init(IOR_param_t *param)
param->testComm),
"Failed to bcast pool info");
if (param->daosStripeCount == -1)
param->daosStripeCount = poolInfo.pi_ntargets * 64UL;
if (o.daosStripeCount == -1)
o.daosStripeCount = poolInfo.pi_ntargets * 64UL;
}
static void DAOS_Fini(IOR_param_t *param)
@ -585,7 +634,7 @@ static void DAOS_Fini(IOR_param_t *param)
int rc;
rc = daos_pool_disconnect(pool, NULL /* ev */);
DCHECK(rc, "Failed to disconnect from pool %s", param->daosPool);
DCHECK(rc, "Failed to disconnect from pool %s", o.daosPool);
rc = daos_eq_destroy(eventQueue, 0 /* flags */);
DCHECK(rc, "Failed to destroy event queue");
@ -612,22 +661,22 @@ static void *DAOS_Open(char *testFileName, IOR_param_t *param)
ghce = fd->containerInfo.ci_epoch_state.es_ghce;
if (param->open == WRITE) {
if (param->daosEpoch == 0)
if (o.daosEpoch == 0)
fd->epoch = ghce + 1;
else if (param->daosEpoch <= ghce)
else if (o.daosEpoch <= ghce)
GERR("Can't modify committed epoch\n");
else
fd->epoch = param->daosEpoch;
fd->epoch = o.daosEpoch;
} else {
if (param->daosEpoch == 0) {
if (param->daosWait == 0)
if (o.daosEpoch == 0) {
if (o.daosWait == 0)
fd->epoch = ghce;
else
fd->epoch = param->daosWait;
} else if (param->daosEpoch > ghce) {
fd->epoch = o.daosWait;
} else if (o.daosEpoch > ghce) {
GERR("Can't read uncommitted epoch\n");
} else {
fd->epoch = param->daosEpoch;
fd->epoch = o.daosEpoch;
}
}
@ -671,15 +720,15 @@ kill_daos_server(IOR_param_t *param)
/* choose the last alive one */
rank = info.pi_ntargets - 1 - info.pi_ndisabled;
rc = uuid_parse(param->daosPool, uuid);
DCHECK(rc, "Failed to parse 'daosPool': %s", param->daosPool);
rc = uuid_parse(o.daosPool, uuid);
DCHECK(rc, "Failed to parse 'daosPool': %s", o.daosPool);
if (rc != 0)
printf("Killing tgt rank: %d (total of %d of %d already disabled)\n",
rank, info.pi_ndisabled, info.pi_ntargets);
fflush(stdout);
rc = daos_mgmt_svc_rip(GetGroup(param), rank, true, NULL);
rc = daos_mgmt_svc_rip(o.daosGroup, rank, true, NULL);
DCHECK(rc, "Error in killing server\n");
targets.rl_nr = 1;
@ -738,13 +787,13 @@ static IOR_offset_t DAOS_Xfer(int access, void *file, IOR_size_t *buffer,
* written
**/
total_size += length;
if (param->daosKill && (access == WRITE) &&
if (o.daosKill && (access == WRITE) &&
((param->blockSize)/2) == total_size) {
/** More than half written lets kill */
if (rank == 0)
printf("Killing and Syncing\n", rank);
kill_and_sync(param);
param->daosKill = 0;
o.daosKill = 0;
}
/*
@ -756,17 +805,17 @@ static IOR_offset_t DAOS_Xfer(int access, void *file, IOR_size_t *buffer,
cfs_list_move_tail(&aio->a_list, &aios);
nAios--;
stripe = (param->offset / param->daosStripeSize) %
param->daosStripeCount;
stripe = (param->offset / o.daosStripeSize) %
o.daosStripeCount;
rc = snprintf(aio->a_dkeyBuf, sizeof aio->a_dkeyBuf, "%lu", stripe);
assert(rc < sizeof aio->a_dkeyBuf);
aio->a_dkey.iov_len = strlen(aio->a_dkeyBuf) + 1;
round = param->offset / (param->daosStripeSize * param->daosStripeCount);
stripeOffset = param->daosStripeSize * round +
param->offset % param->daosStripeSize;
if (param->daosStripeMax != 0)
stripeOffset %= param->daosStripeMax;
aio->a_recx.rx_idx = stripeOffset / param->daosRecordSize;
round = param->offset / (o.daosStripeSize * o.daosStripeCount);
stripeOffset = o.daosStripeSize * round +
param->offset % o.daosStripeSize;
if (o.daosStripeMax != 0)
stripeOffset %= o.daosStripeMax;
aio->a_recx.rx_idx = stripeOffset / o.daosRecordSize;
aio->a_epochRange.epr_lo = fd->epoch;
/*
@ -781,7 +830,7 @@ static IOR_offset_t DAOS_Xfer(int access, void *file, IOR_size_t *buffer,
INFO(VERBOSE_3, param, "Starting AIO %p (%d free %d busy): access %d "
"dkey '%s' iod <%llu, %llu> sgl <%p, %lu>", aio, nAios,
param->daosAios - nAios, access, (char *) aio->a_dkey.iov_buf,
o.daosAios - nAios, access, (char *) aio->a_dkey.iov_buf,
(unsigned long long) aio->a_iod.iod_recxs->rx_idx,
(unsigned long long) aio->a_iod.iod_recxs->rx_nr,
aio->a_sgl.sg_iovs->iov_buf,
@ -805,7 +854,7 @@ static IOR_offset_t DAOS_Xfer(int access, void *file, IOR_size_t *buffer,
* don't have to return valid data as WriteOrRead() doesn't care.
*/
if (access == WRITECHECK || access == READCHECK) {
while (param->daosAios - nAios > 0)
while (o.daosAios - nAios > 0)
AIOWait(param);
memcpy(buffer, aio->a_sgl.sg_iovs->iov_buf, length);
}
@ -818,13 +867,13 @@ static void DAOS_Close(void *file, IOR_param_t *param)
struct fileDescriptor *fd = file;
int rc;
while (param->daosAios - nAios > 0)
while (o.daosAios - nAios > 0)
AIOWait(param);
AIOFini(param);
ObjectClose(fd->object);
if (param->open == WRITE && !param->daosWriteOnly) {
if (param->open == WRITE && !o.daosWriteOnly) {
/* Wait for everybody for to complete the writes. */
MPI_CHECK(MPI_Barrier(param->testComm),
"Failed to synchronize processes");
@ -875,7 +924,7 @@ static char* DAOS_GetVersion()
static void DAOS_Fsync(void *file, IOR_param_t *param)
{
while (param->daosAios - nAios > 0)
while (o.daosAios - nAios > 0)
AIOWait(param);
}

672
src/aiori-DFS.c Executable file
View File

@ -0,0 +1,672 @@
/* -*- 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. *
* *
* Copyright (C) 2018 Intel Corporation
*
* GOVERNMENT LICENSE RIGHTS-OPEN SOURCE SOFTWARE
* The Government's rights to use, modify, reproduce, release, perform, display,
* or disclose this software are subject to the terms of the Apache License as
* provided in Contract No. 8F-30005.
* Any reproduction of computer software, computer software documentation, or
* portions thereof marked with this legend must also reproduce the markings.
********************************************************************************
*
* Implement of abstract I/O interface for DFS.
*
\******************************************************************************/
#ifdef HAVE_CONFIG_H
#include "config.h"
#endif
#include <string.h>
#include <errno.h>
#include <stdio.h>
#include <dirent.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <unistd.h>
#include <fcntl.h>
#include <libgen.h>
#include <daos_types.h>
#include <daos_api.h>
#include <daos_fs.h>
#include "ior.h"
#include "aiori.h"
#include "iordef.h"
#include "utilities.h"
dfs_t *dfs;
daos_handle_t poh, coh;
/************************** O P T I O N S *****************************/
struct dfs_options{
char * pool;
char * svcl;
char * group;
char * cont;
};
static struct dfs_options o = {
.pool = NULL,
.svcl = NULL,
.group = NULL,
.cont = NULL,
};
static option_help options [] = {
{'p', "pool", "DAOS pool uuid", OPTION_REQUIRED_ARGUMENT, 's', & o.pool},
{'s', "svcl", "DAOS pool SVCL", OPTION_REQUIRED_ARGUMENT, 's', & o.svcl},
{'g', "group", "DAOS server group", OPTION_OPTIONAL_ARGUMENT, 's', & o.group},
{'c', "cont", "DFS container uuid", OPTION_REQUIRED_ARGUMENT, 's', & o.cont},
LAST_OPTION
};
/**************************** 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 char* DFS_GetVersion();
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 *);
static void DFS_Init(IOR_param_t *param);
static void DFS_Finalize(IOR_param_t *param);
static option_help * DFS_options();
/************************** 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,
.get_version = DFS_GetVersion,
.fsync = DFS_Fsync,
.get_file_size = DFS_GetFileSize,
.statfs = DFS_Statfs,
.mkdir = DFS_Mkdir,
.rmdir = DFS_Rmdir,
.access = DFS_Access,
.stat = DFS_Stat,
.initialize = DFS_Init,
.finalize = DFS_Finalize,
.get_options = DFS_options,
};
/***************************** F U N C T I O N S ******************************/
/* For DAOS methods. */
#define DCHECK(rc, format, ...) \
do { \
int _rc = (rc); \
\
if (_rc < 0) { \
fprintf(stderr, "ERROR (%s:%d): %d: %d: " \
format"\n", __FILE__, __LINE__, rank, _rc, \
##__VA_ARGS__); \
fflush(stderr); \
MPI_Abort(MPI_COMM_WORLD, -1); \
} \
} while (0)
#define DERR(rc, format, ...) \
do { \
int _rc = (rc); \
\
if (_rc < 0) { \
fprintf(stderr, "ERROR (%s:%d): %d: %d: " \
format"\n", __FILE__, __LINE__, rank, _rc, \
##__VA_ARGS__); \
fflush(stderr); \
goto out; \
} \
} while (0)
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[1024];
if (getcwd(cwd, 1024) == NULL)
D_GOTO(out, rc = -ENOMEM);
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;
}
static option_help * DFS_options(){
return options;
}
static void
DFS_Init(IOR_param_t *param) {
uuid_t pool_uuid, co_uuid;
daos_pool_info_t pool_info;
daos_cont_info_t co_info;
d_rank_list_t *svcl = NULL;
bool cont_created = false;
int rc;
if (o.pool == NULL || o.svcl == NULL || o.cont == NULL)
ERR("Invalid Arguments to DFS\n");
rc = uuid_parse(o.pool, pool_uuid);
DCHECK(rc, "Failed to parse 'Pool uuid': %s", o.pool);
rc = uuid_parse(o.cont, co_uuid);
DCHECK(rc, "Failed to parse 'Cont uuid': %s", o.cont);
svcl = daos_rank_list_parse(o.svcl, ":");
if (svcl == NULL)
ERR("Failed to allocate svcl");
if (verbose >= 3) {
printf("Pool uuid = %s, SVCL = %s\n", o.pool, o.svcl);
printf("DFS Container namespace uuid = %s\n", o.cont);
}
rc = daos_init();
DCHECK(rc, "Failed to initialize daos");
/** Connect to DAOS pool */
rc = daos_pool_connect(pool_uuid, o.group, svcl, DAOS_PC_RW, &poh,
&pool_info, NULL);
DCHECK(rc, "Failed to connect to pool");
rc = daos_cont_open(poh, co_uuid, DAOS_COO_RW, &coh, &co_info, NULL);
/* If NOEXIST we create it */
if (rc == -DER_NONEXIST) {
if (verbose >= 3)
printf("Creating DFS Container ...\n");
rc = daos_cont_create(poh, co_uuid, NULL);
if (rc == 0) {
cont_created = true;
rc = daos_cont_open(poh, co_uuid, DAOS_COO_RW, &coh,
&co_info, NULL);
}
}
DCHECK(rc, "Failed to create container");
rc = dfs_mount(poh, coh, O_RDWR, &dfs);
DCHECK(rc, "Failed to mount DFS namespace");
}
static void
DFS_Finalize(IOR_param_t *param)
{
int rc;
rc = dfs_umount(dfs, true);
DCHECK(rc, "Failed to umount DFS namespace");
rc = daos_cont_close(coh, NULL);
DCHECK(rc, "Failed to close container");
daos_pool_disconnect(poh, NULL);
DCHECK(rc, "Failed to disconnect from pool");
rc = daos_fini();
DCHECK(rc, "Failed to finalize DAOS");
}
/*
* 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, mode;
int fd_oflag = 0;
int rc;
assert(param);
fd_oflag |= O_CREAT | O_RDWR;
mode = S_IFREG | param->mode;
rc = parse_filename(testFileName, &name, &dir_name);
DERR(rc, "Failed to parse path %s", testFileName);
assert(dir_name);
assert(name);
rc = dfs_lookup(dfs, dir_name, O_RDWR, &parent, &pmode);
DERR(rc, "dfs_lookup() of %s Failed", dir_name);
rc = dfs_open(dfs, parent, name, mode, fd_oflag, DAOS_OC_LARGE_RW,
NULL, &obj);
DERR(rc, "dfs_open() of %s Failed", name);
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);
DERR(rc, "Failed to parse path %s", testFileName);
assert(dir_name);
assert(name);
rc = dfs_lookup(dfs, dir_name, O_RDWR, &parent, &pmode);
DERR(rc, "dfs_lookup() of %s Failed", dir_name);
rc = dfs_open(dfs, parent, name, S_IFREG, fd_oflag, 0, NULL, &obj);
DERR(rc, "dfs_open() of %s Failed", name);
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) {
fprintf(stderr, "dfs_write() failed (%d)", rc);
return -1;
}
ret = remaining;
} else {
rc = dfs_read(dfs, obj, sgl, param->offset, &ret);
if (rc || ret == 0)
fprintf(stderr, "dfs_read() failed(%d)", rc);
}
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)
{
dfs_sync(dfs);
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);
DERR(rc, "Failed to parse path %s", testFileName);
assert(dir_name);
assert(name);
rc = dfs_lookup(dfs, dir_name, O_RDWR, &parent, &pmode);
DERR(rc, "dfs_lookup() of %s Failed", dir_name);
rc = dfs_remove(dfs, parent, name, false);
DERR(rc, "dfs_remove() of %s Failed", name);
out:
if (name)
free(name);
if (dir_name)
free(dir_name);
if (parent)
dfs_release(parent);
}
static char* DFS_GetVersion()
{
static char ver[1024] = {};
sprintf(ver, "%s", "DAOS");
return ver;
}
/*
* 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, O_RDONLY, &obj, NULL);
if (rc) {
fprintf(stderr, "dfs_lookup() of %s Failed (%d)", testFileName, 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);
DERR(rc, "Failed to parse path %s", path);
assert(dir_name);
assert(name);
rc = dfs_lookup(dfs, dir_name, O_RDWR, &parent, &pmode);
DERR(rc, "dfs_lookup() of %s Failed", dir_name);
rc = dfs_mkdir(dfs, parent, name, mode);
DERR(rc, "dfs_mkdir() of %s Failed", name);
out:
if (name)
free(name);
if (dir_name)
free(dir_name);
if (parent)
dfs_release(parent);
if (rc)
return -1;
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);
DERR(rc, "Failed to parse path %s", path);
assert(dir_name);
assert(name);
rc = dfs_lookup(dfs, dir_name, O_RDWR, &parent, &pmode);
DERR(rc, "dfs_lookup() of %s Failed", dir_name);
rc = dfs_remove(dfs, parent, name, false);
DERR(rc, "dfs_remove() of %s Failed", name);
out:
if (name)
free(name);
if (dir_name)
free(dir_name);
if (parent)
dfs_release(parent);
if (rc)
return -1;
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);
DERR(rc, "Failed to parse path %s", path);
assert(dir_name);
rc = dfs_lookup(dfs, dir_name, O_RDWR, &parent, &pmode);
DERR(rc, "dfs_lookup() of %s Failed", dir_name);
if (name && strcmp(name, ".") == 0) {
free(name);
name = NULL;
}
rc = dfs_stat(dfs, parent, name, &stbuf);
DERR(rc, "dfs_stat() of %s Failed", name);
out:
if (name)
free(name);
if (dir_name)
free(dir_name);
if (parent)
dfs_release(parent);
if (rc)
return -1;
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);
DERR(rc, "Failed to parse path %s", path);
assert(dir_name);
assert(name);
rc = dfs_lookup(dfs, dir_name, O_RDONLY, &parent, &pmode);
DERR(rc, "dfs_lookup() of %s Failed", dir_name);
rc = dfs_stat(dfs, parent, name, buf);
DERR(rc, "dfs_stat() of %s Failed", name);
out:
if (name)
free(name);
if (dir_name)
free(dir_name);
if (parent)
dfs_release(parent);
if (rc)
return -1;
return rc;
}

1
src/aiori.c
View File

@ -60,6 +60,7 @@ ior_aiori_t *available_aiori[] = {
#endif
#ifdef USE_DAOS_AIORI
&daos_aiori,
&dfs_aiori,
#endif
NULL
};

1
src/aiori.h
View File

@ -97,6 +97,7 @@ extern ior_aiori_t s3_plus_aiori;
extern ior_aiori_t s3_emc_aiori;
extern ior_aiori_t rados_aiori;
extern ior_aiori_t daos_aiori;
extern ior_aiori_t dfs_aiori;
void aiori_initialize(IOR_test_t *th);
void aiori_finalize(IOR_test_t *th);

3
src/ior-output.c
View File

@ -315,7 +315,8 @@ void ShowTestStart(IOR_param_t *test)
PrintKeyValInt("TestID", test->id);
PrintKeyVal("StartTime", CurrentTimeString());
/* if pvfs2:, then skip */
if (Regex(test->testFileName, "^[a-z][a-z].*:") == 0) {
if (strcasecmp(test->api, "DFS") &&
Regex(test->testFileName, "^[a-z][a-z].*:") == 0) {
DisplayFreespace(test);
}

15
src/ior.c
View File

@ -123,8 +123,6 @@ int ior_main(int argc, char **argv)
PrintHeader(argc, argv);
aiori_initialize(tests_head);
/* perform each test */
for (tptr = tests_head; tptr != NULL; tptr = tptr->next) {
verbose = tptr->params.verbose;
@ -139,12 +137,11 @@ int ior_main(int argc, char **argv)
sleep(5);
fprintf(out_logfile, "\trank %d: awake.\n", rank);
}
TestIoSys(tptr);
ShowTestEnd(tptr);
}
aiori_finalize(tests_head);
if (verbose < 0)
/* always print final summary */
verbose = 0;
@ -199,11 +196,6 @@ void init_IOR_Param_t(IOR_param_t * p)
p->setAlignment = 1;
p->lustre_start_ost = -1;
p->daosRecordSize = 262144;
p->daosStripeSize = 524288;
p->daosStripeCount = -1;
p->daosAios = 1;
hdfs_user = getenv("USER");
if (!hdfs_user)
hdfs_user = "";
@ -1194,6 +1186,9 @@ static void TestIoSys(IOR_test_t *test)
/* bind I/O calls to specific API */
backend = aiori_select(params->api);
if (backend->initialize)
backend->initialize(params);
/* show test setup */
if (rank == 0 && verbose >= VERBOSE_0)
ShowSetup(params);
@ -1478,6 +1473,8 @@ static void TestIoSys(IOR_test_t *test)
free(timer[i]);
}
if (backend->finalize)
backend->finalize(NULL);
/* Sync with the tasks that did not participate in this test */
MPI_CHECK(MPI_Barrier(mpi_comm_world), "barrier error");

15
src/ior.h
View File

@ -190,21 +190,6 @@ typedef struct
int lustre_set_striping; /* flag that we need to set lustre striping */
int lustre_ignore_locks;
/* DAOS variables */
char daosGroup[MAX_STR]; /* group name */
char daosPool[37]; /* pool UUID */
char daosPoolSvc[MAX_STR]; /* pool service ranks */
int daosRecordSize; /* size of akey record (i.e., rx_rsize) */
int daosStripeSize;
unsigned long daosStripeCount;
unsigned long daosStripeMax; /* max length of a stripe */
int daosAios; /* max number of concurrent async I/Os */
int daosWriteOnly; /* write only, no flush and commit */
unsigned long daosEpoch; /* epoch to access */
unsigned long daosWait; /* epoch to wait for before reading */
int daosKill; /* kill a target while running IOR */
char daosObjectClass[MAX_STR]; /* object class */
/* gpfs variables */
int gpfs_hint_access; /* use gpfs "access range" hint */
int gpfs_release_token; /* immediately release GPFS tokens after

2
src/mdtest-main.c
View File

@ -3,11 +3,9 @@
int main(int argc, char **argv) {
MPI_Init(&argc, &argv);
aiori_initialize(NULL);
mdtest_run(argc, argv, MPI_COMM_WORLD, stdout);
aiori_finalize(NULL);
MPI_Finalize();
return 0;
}

15
src/mdtest.c
View File

@ -1615,8 +1615,9 @@ void valid_tests() {
FAIL("-c not compatible with -B");
}
if ( strcasecmp(backend_name, "POSIX") != 0 && strcasecmp(backend_name, "DUMMY") != 0) {
FAIL("-a only supported interface is POSIX (and DUMMY) right now!");
if ( strcasecmp(backend_name, "POSIX") != 0 && strcasecmp(backend_name, "DUMMY") != 0 &&
strcasecmp(backend_name, "DFS") != 0) {
FAIL("-a only supported interface is POSIX, DFS and DUMMY right now!");
}
/* check for shared file incompatibilities */
@ -1757,6 +1758,9 @@ void display_freespace(char *testdirpath)
strcpy(dirpath, ".");
}
if (strcasecmp(backend_name, "DFS") == 0)
return;
if (verbose >= 3 && rank == 0) {
fprintf(out_logfile, "V-3: Before show_file_system_size, dirpath is \"%s\"\n", dirpath );
fflush( out_logfile );
@ -2223,6 +2227,9 @@ mdtest_results_t * mdtest_run(int argc, char **argv, MPI_Comm world_com, FILE *
option_parse(argc - parsed_options, argv + parsed_options, backend->get_options(), & printhelp);
}
if (backend->initialize)
backend->initialize(NULL);
if(printhelp != 0){
printf("Usage: %s ", argv[0]);
@ -2517,5 +2524,9 @@ mdtest_results_t * mdtest_run(int argc, char **argv, MPI_Comm world_com, FILE *
if (random_seed > 0) {
free(rand_array);
}
if (backend->finalize)
backend->finalize(NULL);
return summary_table;
}

12
src/option.c
View File

@ -89,6 +89,10 @@ static int print_value(option_help * o){
pos += printf("=%lld", *(long long*) o->variable);
break;
}
case('u'):{
pos += printf("=%lu", *(uint64_t*) o->variable);
break;
}
}
}
if (o->arg == OPTION_FLAG && (*(int*)o->variable) != 0){
@ -213,6 +217,10 @@ static int print_option_value(option_help * o){
pos += printf("=%lld", *(long long*) o->variable);
break;
}
case('u'):{
pos += printf("=%lu", *(uint64_t*) o->variable);
break;
}
}
}else{
//printf(" ");
@ -354,6 +362,10 @@ int option_parse(int argc, char ** argv, option_help * args, int * printhelp){
*(long long*) o->variable = string_to_bytes(arg);
break;
}
case('u'):{
*(uint64_t*) o->variable = string_to_bytes(arg);
break;
}
default:
printf("ERROR: Unknown option type %c\n", o->type);
}

26
src/parse_options.c
View File

@ -300,32 +300,6 @@ void DecodeDirective(char *line, IOR_param_t *params)
params->numTasks = atoi(value);
} else if (strcasecmp(option, "summaryalways") == 0) {
params->summary_every_test = atoi(value);
} else if (strcasecmp(option, "daosgroup") == 0) {
strcpy(params->daosGroup, value);
} else if (strcasecmp(option, "daospool") == 0) {
strcpy(params->daosPool, value);
} else if (strcasecmp(option, "daospoolsvc") == 0) {
strcpy(params->daosPoolSvc, value);
} else if (strcasecmp(option, "daosrecordsize") == 0) {
params->daosRecordSize = string_to_bytes(value);
} else if (strcasecmp(option, "daosstripesize") == 0) {
params->daosStripeSize = string_to_bytes(value);
} else if (strcasecmp(option, "daosstripecount") == 0) {
params->daosStripeCount = atoi(value);
} else if (strcasecmp(option, "daosstripemax") == 0) {
params->daosStripeMax = string_to_bytes(value);
} else if (strcasecmp(option, "daosaios") == 0) {
params->daosAios = atoi(value);
} else if (strcasecmp(option, "daosepoch") == 0) {
params->daosEpoch = atoi(value);
} else if (strcasecmp(option, "daoswait") == 0) {
params->daosWait = atoi(value);
} else if (strcasecmp(option, "daoswriteonly") == 0) {
params->daosWriteOnly = atoi(value);
} else if (strcasecmp(option, "daoskill") == 0) {
params->daosKill = atoi(value);
} else if (strcasecmp(option, "daosobjectclass") == 0) {
strcpy(params->daosObjectClass, value);
} else {
if (rank == 0)
fprintf(out_logfile, "Unrecognized parameter \"%s\"\n",