Loading...
Searching...
No Matches
XrdCephPosix.cc
Go to the documentation of this file.
//------------------------------------------------------------------------------
// Copyright (c) 2014-2015 by European Organization for Nuclear Research (CERN)
// Author: Sebastien Ponce <sebastien.ponce@cern.ch>
//------------------------------------------------------------------------------
// This file is part of the XRootD software suite.
//
// XRootD is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// XRootD is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with XRootD. If not, see <http://www.gnu.org/licenses/>.
//
// In applying this licence, CERN does not waive the privileges and immunities
// granted to it by virtue of its status as an Intergovernmental Organization
// or submit itself to any jurisdiction.
//------------------------------------------------------------------------------
/*
* This interface provides wrapper methods for using ceph through a POSIX API.
*/
#include <sys/types.h>
#include <sys/stat.h>
#include <cerrno>
#include <fcntl.h>
#include <unistd.h>
#include <stdlib.h>
#include <stdarg.h>
#include <memory>
#include <radosstriper/libradosstriper.hpp>
#include <map>
#include <stdexcept>
#include <string>
#include <sstream>
#include <sys/xattr.h>
#include <time.h>
#include <chrono>
#include <limits>
#include <pthread.h>
#include "XrdSfs/XrdSfsAio.hh"
#include "XrdSys/XrdSysPthread.hh"
#include "XrdOuc/XrdOucName2Name.hh"
#include "XrdSys/XrdSysPlatform.hh"
#include <XrdOss/XrdOss.hh>
#include "XrdOuc/XrdOucIOVec.hh"
#include "XrdCeph/XrdCephPosix.hh"
#include "XrdCeph/XrdCephBulkAioRead.hh"
#include "XrdSfs/XrdSfsFlags.hh" // for the OFFLINE flag status
struct DirIterator {
librados::NObjectIterator m_iterator;
librados::IoCtx *m_ioctx;
};
struct AioArgs {
AioArgs(XrdSfsAio* a, AioCB *b, size_t n, int _fd, ceph::bufferlist *_bl=0) :
aiop(a), callback(b), nbBytes(n), fd(_fd), bl(_bl) { ::gettimeofday(&startTime, nullptr); }
XrdSfsAio* aiop;
AioCB *callback;
size_t nbBytes;
int fd;
::timeval startTime;
ceph::bufferlist *bl;
};
typedef std::map<std::string, libradosstriper::RadosStriper*> StriperDict;
std::vector<StriperDict> g_radosStripers;
typedef std::map<std::string, librados::IoCtx*> IOCtxDict;
std::vector<IOCtxDict> g_ioCtx;
std::vector<librados::Rados*> g_cluster;
XrdSysMutex g_striper_mutex;
unsigned int g_cephPoolIdx = 0;
unsigned int g_cephAioWaitThresh = 15;
unsigned int g_maxCephPoolIdx = 1;
XrdOucName2Name *g_namelib = 0;
std::multiset<std::string> g_filesOpenForWrite;
std::map<unsigned int, CephFileRef> g_fds;
unsigned int g_nextCephFd = 0;
XrdSysMutex g_fd_mutex;
XrdSysMutex g_init_mutex;
//JW Counter for number of times a given cluster is resolved.
std::map<unsigned int, unsigned long long> g_idxCntr;
unsigned int getCephPoolIdxAndIncrease() {
if (g_radosStripers.size() == 0) {
// make sure we do not have a race condition here
XrdSysMutexHelper lock(g_init_mutex);
// double check now that we have the lock
if (g_radosStripers.size() == 0) {
// initialization phase : allocate corresponding places in the vectors
for (unsigned int i = 0; i < g_maxCephPoolIdx; i++) {
g_radosStripers.push_back(StriperDict());
g_ioCtx.push_back(IOCtxDict());
g_cluster.push_back(0);
}
}
}
unsigned int res = g_cephPoolIdx;
unsigned nextValue = g_cephPoolIdx+1;
if (nextValue >= g_maxCephPoolIdx) {
nextValue = 0;
}
g_cephPoolIdx = nextValue;
// JW logging of accesses:
++g_idxCntr[res];
return res;
}
bool isOpenForWrite(std::string& name) {
XrdSysMutexHelper lock(g_fd_mutex);
return g_filesOpenForWrite.find(name) != g_filesOpenForWrite.end();
}
CephFileRef* getFileRef(int fd) {
XrdSysMutexHelper lock(g_fd_mutex);
std::map<unsigned int, CephFileRef>::iterator it = g_fds.find(fd);
if (it != g_fds.end()) {
// We will release the lock upon exiting this function.
// The structure here is not protected from deletion, but we trust xrootd to
// ensure close (which does the deletion) will not be called before all previous
// calls are complete (including the async ones).
return &(it->second);
} else {
return 0;
}
}
void deleteFileRef(int fd, const CephFileRef &fr) {
XrdSysMutexHelper lock(g_fd_mutex);
if ((fr.flags & O_ACCMODE) != O_RDONLY) {
g_filesOpenForWrite.erase(g_filesOpenForWrite.find(fr.name));
}
std::map<unsigned int, CephFileRef>::iterator it = g_fds.find(fd);
if (it != g_fds.end()) {
g_fds.erase(it);
}
}
int insertFileRef(CephFileRef &fr) {
XrdSysMutexHelper lock(g_fd_mutex);
g_fds[g_nextCephFd] = fr;
g_nextCephFd++;
if ((fr.flags & O_ACCMODE) != O_RDONLY) {
g_filesOpenForWrite.insert(fr.name);
}
return g_nextCephFd-1;
}
CephFile g_defaultParams = { "",
"default", // default pool
"admin", // default user
1, // default nbStripes
4 * 1024 * 1024, // default stripeUnit : 4 MB
4 * 1024 * 1024}; // default objectSize : 4 MB
std::string g_defaultUserId = "admin";
std::string g_defaultPool = "default";
static void (*g_logfunc) (char *, va_list argp) = 0;
static void logwrapper(char* format, ...) {
if (0 == g_logfunc) return;
va_list arg;
va_start(arg, format);
(*g_logfunc)(format, arg);
va_end(arg);
}
static unsigned long long int stoull(const std::string &s) {
char* end;
errno = 0;
unsigned long long int res = strtoull(s.c_str(), &end, 10);
if (0 != *end) {
throw std::invalid_argument(s);
}
if (ERANGE == errno) {
throw std::out_of_range(s);
}
return res;
}
static unsigned int stoui(const std::string &s) {
char* end;
errno = 0;
unsigned long int res = strtoul(s.c_str(), &end, 10);
if (0 != *end) {
throw std::invalid_argument(s);
}
if (ERANGE == errno || res > std::numeric_limits<unsigned int>::max()) {
throw std::out_of_range(s);
}
return (unsigned int)res;
}
void dumpClusterInfo() {
//JW
// log the current state of the cluster:
// don't want to lock here, so the numbers may not be 100% self-consistent
int n_cluster = g_cluster.size();
int n_ioCtx = g_ioCtx.size();
int n_filesOpenForWrite = g_filesOpenForWrite.size();
int n_fds = g_fds.size();
int n_stripers = g_radosStripers.size();
int n_stripers_pool = 0;
for (size_t i = 0; i < g_radosStripers.size(); ++i) {
n_stripers_pool += g_radosStripers.at(i).size();
}
std::stringstream ss;
ss << "Counts: " << n_cluster << " " << n_ioCtx << " " << n_filesOpenForWrite << " "
<< n_fds << " " << n_stripers << " " << n_stripers_pool << " " << n_stripers_pool
<< " CountsbyCluster: [";
for (const auto& el : g_idxCntr) {
ss << el.first << ":" << el.second << ", " ;
} // it
ss<< "], ";
logwrapper((char*)"dumpClusterInfo : %s", ss.str().c_str());
}
static int fillCephUserId(const std::string &params, XrdOucEnv *env, CephFile &file) {
// default
file.userId = g_defaultParams.userId;
// parsing
size_t atPos = params.find('@');
if (std::string::npos != atPos) {
file.userId = params.substr(0, atPos);
return atPos+1;
} else {
if (0 != env) {
char* cuser = env->Get("cephUserId");
if (0 != cuser) {
file.userId = cuser;
}
}
return 0;
}
}
static int fillCephPool(const std::string &params, unsigned int offset, XrdOucEnv *env, CephFile &file) {
// default
file.pool = g_defaultParams.pool;
// parsing
size_t comPos = params.find(',', offset);
if (std::string::npos == comPos) {
if (params.size() == offset) {
if (NULL != env) {
char* cpool = env->Get("cephPool");
if (0 != cpool) {
file.pool = cpool;
}
}
} else {
file.pool = params.substr(offset);
}
return params.size();
} else {
file.pool = params.substr(offset, comPos-offset);
return comPos+1;
}
}
// this may raise std::invalid_argument and std::out_of_range
static int fillCephNbStripes(const std::string &params, unsigned int offset, XrdOucEnv *env, CephFile &file) {
// default
file.nbStripes = g_defaultParams.nbStripes;
// parsing
size_t comPos = params.find(',', offset);
if (std::string::npos == comPos) {
if (params.size() == offset) {
if (NULL != env) {
char* cNbStripes = env->Get("cephNbStripes");
if (0 != cNbStripes) {
file.nbStripes = stoui(cNbStripes);
}
}
} else {
file.nbStripes = stoui(params.substr(offset));
}
return params.size();
} else {
file.nbStripes = stoui(params.substr(offset, comPos-offset));
return comPos+1;
}
}
// this may raise std::invalid_argument and std::out_of_range
static int fillCephStripeUnit(const std::string &params, unsigned int offset, XrdOucEnv *env, CephFile &file) {
// default
file.stripeUnit = g_defaultParams.stripeUnit;
// parsing
size_t comPos = params.find(',', offset);
if (std::string::npos == comPos) {
if (params.size() == offset) {
if (NULL != env) {
char* cStripeUnit = env->Get("cephStripeUnit");
if (0 != cStripeUnit) {
file.stripeUnit = ::stoull(cStripeUnit);
}
}
} else {
file.stripeUnit = ::stoull(params.substr(offset));
}
return params.size();
} else {
file.stripeUnit = ::stoull(params.substr(offset, comPos-offset));
return comPos+1;
}
}
// this may raise std::invalid_argument and std::out_of_range
static void fillCephObjectSize(const std::string &params, unsigned int offset, XrdOucEnv *env, CephFile &file) {
// default
file.objectSize = g_defaultParams.objectSize;
// parsing
if (params.size() == offset) {
if (NULL != env) {
char* cObjectSize = env->Get("cephObjectSize");
if (0 != cObjectSize) {
file.objectSize = ::stoull(cObjectSize);
}
}
} else {
file.objectSize = ::stoull(params.substr(offset));
}
}
void fillCephFileParams(const std::string &params, XrdOucEnv *env, CephFile &file) {
// parse the params one by one
unsigned int afterUser = fillCephUserId(params, env, file);
unsigned int afterPool = fillCephPool(params, afterUser, env, file);
unsigned int afterNbStripes = fillCephNbStripes(params, afterPool, env, file);
unsigned int afterStripeUnit = fillCephStripeUnit(params, afterNbStripes, env, file);
fillCephObjectSize(params, afterStripeUnit, env, file);
}
void ceph_posix_set_defaults(const char* value) {
if (value) {
CephFile newdefault;
fillCephFileParams(value, NULL, newdefault);
g_defaultParams = newdefault;
}
}
void translateFileName(std::string &physName, std::string logName){
if (0 != g_namelib) {
char physCName[MAXPATHLEN+1];
int retc = g_namelib->lfn2pfn(logName.c_str(), physCName, sizeof(physCName));
if (retc) {
logwrapper((char*)"ceph_namelib : failed to translate %s using namelib plugin, using it as is", logName.c_str());
physName = logName;
} else {
logwrapper((char*)"ceph_namelib : translated %s to %s", logName.c_str(), physCName);
physName = physCName;
}
} else {
//logwrapper((char*)"ceph_namelib : No mapping done");
physName = logName;
}
}
void fillCephFile(const char *path, XrdOucEnv *env, CephFile &file) {
// Syntax of the given path is :
// [[userId@]pool[,nbStripes[,stripeUnit[,objectSize]]]:]<actual path>
// for the missing parts, if env is not null the entries
// cephUserId, cephPool, cephNbStripes, cephStripeUnit, and cephObjectSize
// of env will be used.
// If env is null or no entry is found for what is missing, defaults are
// applied. These defaults are initially set to 'admin', 'default', 1, 4MB and 4MB
// but can be changed via a call to ceph_posix_set_defaults
std::string spath {path};
// If namelib is specified, apply translation to the whole path (which might include pool, etc)
translateFileName(spath,path);
size_t colonPos = spath.find(':');
if (std::string::npos == colonPos) {
// deal with name translation
file.name = spath;
fillCephFileParams("", env, file);
} else {
file.name = spath.substr(colonPos+1);
fillCephFileParams(spath.substr(0, colonPos), env, file);
}
}
static CephFile getCephFile(const char *path, XrdOucEnv *env) {
CephFile file;
fillCephFile(path, env, file);
return file;
}
static CephFileRef getCephFileRef(const char *path, XrdOucEnv *env, int flags,
mode_t mode, unsigned long long offset) {
CephFileRef fr;
fillCephFile(path, env, fr);
fr.flags = flags;
fr.mode = mode;
fr.offset = 0;
fr.maxOffsetWritten = 0;
fr.bytesAsyncWritePending = 0;
fr.bytesWritten = 0;
fr.rdcount = 0;
fr.wrcount = 0;
fr.asyncRdStartCount = 0;
fr.asyncRdCompletionCount = 0;
fr.asyncWrStartCount = 0;
fr.asyncWrCompletionCount = 0;
fr.lastAsyncSubmission.tv_sec = 0;
fr.lastAsyncSubmission.tv_usec = 0;
fr.longestAsyncWriteTime = 0.0l;
fr.longestCallbackInvocation = 0.0l;
return fr;
}
inline librados::Rados* checkAndCreateCluster(unsigned int cephPoolIdx,
std::string userId = g_defaultParams.userId) {
if (0 == g_cluster[cephPoolIdx]) {
// create connection to cluster
librados::Rados *cluster = new librados::Rados;
if (0 == cluster) {
return 0;
}
int rc = cluster->init(userId.c_str());
if (rc) {
logwrapper((char*)"checkAndCreateCluster : cluster init failed");
delete cluster;
return 0;
}
rc = cluster->conf_read_file(NULL);
if (rc) {
logwrapper((char*)"checkAndCreateCluster : cluster read config failed, rc = %d", rc);
cluster->shutdown();
delete cluster;
return 0;
}
cluster->conf_parse_env(NULL);
rc = cluster->connect();
if (rc) {
logwrapper((char*)"checkAndCreateCluster : cluster connect failed, rc = %d", rc);
cluster->shutdown();
delete cluster;
return 0;
}
g_cluster[cephPoolIdx] = cluster;
}
return g_cluster[cephPoolIdx];
}
int checkAndCreateStriper(unsigned int cephPoolIdx, std::string &userAtPool, const CephFile& file) {
StriperDict &sDict = g_radosStripers[cephPoolIdx];
StriperDict::iterator it = sDict.find(userAtPool);
if (it == sDict.end()) {
// we need to create a new radosStriper
// Get a cluster
librados::Rados* cluster = checkAndCreateCluster(cephPoolIdx, file.userId);
if (0 == cluster) {
logwrapper((char*)"checkAndCreateStriper : checkAndCreateCluster failed");
return 0;
}
// create IoCtx for our pool
librados::IoCtx *ioctx = new librados::IoCtx;
if (0 == ioctx) {
logwrapper((char*)"checkAndCreateStriper : IoCtx instantiation failed");
cluster->shutdown();
delete cluster;
g_cluster[cephPoolIdx] = 0;
return 0;
}
int rc = g_cluster[cephPoolIdx]->ioctx_create(file.pool.c_str(), *ioctx);
if (rc != 0) {
logwrapper((char*)"checkAndCreateStriper : ioctx_create failed, rc = %d", rc);
cluster->shutdown();
delete cluster;
g_cluster[cephPoolIdx] = 0;
delete ioctx;
return 0;
}
// create RadosStriper connection
libradosstriper::RadosStriper *striper = new libradosstriper::RadosStriper;
if (0 == striper) {
logwrapper((char*)"checkAndCreateStriper : RadosStriper instantiation failed");
delete ioctx;
cluster->shutdown();
delete cluster;
g_cluster[cephPoolIdx] = 0;
return 0;
}
rc = libradosstriper::RadosStriper::striper_create(*ioctx, striper);
if (rc != 0) {
logwrapper((char*)"checkAndCreateStriper : striper_create failed, rc = %d", rc);
delete striper;
delete ioctx;
cluster->shutdown();
delete cluster;
g_cluster[cephPoolIdx] = 0;
return 0;
}
// setup layout
rc = striper->set_object_layout_stripe_count(file.nbStripes);
if (rc != 0) {
logwrapper((char*)"checkAndCreateStriper : invalid nbStripes %d", file.nbStripes);
delete striper;
delete ioctx;
cluster->shutdown();
delete cluster;
g_cluster[cephPoolIdx] = 0;
return 0;
}
rc = striper->set_object_layout_stripe_unit(file.stripeUnit);
if (rc != 0) {
logwrapper((char*)"checkAndCreateStriper : invalid stripeUnit %d (must be non 0, multiple of 64K)", file.stripeUnit);
delete striper;
delete ioctx;
cluster->shutdown();
delete cluster;
g_cluster[cephPoolIdx] = 0;
return 0;
}
rc = striper->set_object_layout_object_size(file.objectSize);
if (rc != 0) {
logwrapper((char*)"checkAndCreateStriper : invalid objectSize %d (must be non 0, multiple of stripe_unit)", file.objectSize);
delete striper;
delete ioctx;
cluster->shutdown();
delete cluster;
g_cluster[cephPoolIdx] = 0;
return 0;
}
IOCtxDict & ioDict = g_ioCtx[cephPoolIdx];
ioDict.emplace(userAtPool, ioctx);
sDict.emplace(userAtPool, striper);
}
return 1;
}
static libradosstriper::RadosStriper* getRadosStriper(const CephFile& file) {
XrdSysMutexHelper lock(g_striper_mutex);
std::stringstream ss;
ss << file.userId << '@' << file.pool << ',' << file.nbStripes << ','
<< file.stripeUnit << ',' << file.objectSize;
std::string userAtPool = ss.str();
unsigned int cephPoolIdx = getCephPoolIdxAndIncrease();
if (checkAndCreateStriper(cephPoolIdx, userAtPool, file) == 0) {
logwrapper((char*)"getRadosStriper : checkAndCreateStriper failed");
return 0;
}
return g_radosStripers[cephPoolIdx][userAtPool];
}
static librados::IoCtx* getIoCtx(const CephFile& file) {
XrdSysMutexHelper lock(g_striper_mutex);
std::stringstream ss;
ss << file.userId << '@' << file.pool << ',' << file.nbStripes << ','
<< file.stripeUnit << ',' << file.objectSize;
std::string userAtPool = ss.str();
unsigned int cephPoolIdx = getCephPoolIdxAndIncrease();
if (checkAndCreateStriper(cephPoolIdx, userAtPool, file) == 0) {
return 0;
}
return g_ioCtx[cephPoolIdx][userAtPool];
}
void ceph_posix_disconnect_all() {
XrdSysMutexHelper lock(g_striper_mutex);
for (unsigned int i= 0; i < g_maxCephPoolIdx; i++) {
for (StriperDict::iterator it2 = g_radosStripers[i].begin();
it2 != g_radosStripers[i].end();
it2++) {
delete it2->second;
}
for (IOCtxDict::iterator it2 = g_ioCtx[i].begin();
it2 != g_ioCtx[i].end();
it2++) {
delete it2->second;
}
delete g_cluster[i];
}
g_radosStripers.clear();
g_ioCtx.clear();
g_cluster.clear();
}
void ceph_posix_set_logfunc(void (*logfunc) (char *, va_list argp)) {
g_logfunc = logfunc;
};
static int ceph_posix_internal_truncate(const CephFile &file, unsigned long long size);
int ceph_posix_open(XrdOucEnv* env, const char *pathname, int flags, mode_t mode){
CephFileRef fr = getCephFileRef(pathname, env, flags, mode, 0);
struct stat buf;
libradosstriper::RadosStriper *striper = getRadosStriper(fr); //Get a handle to the RADOS striper API
if (NULL == striper) {
logwrapper((char*)"Cannot create striper");
return -EINVAL;
}
dumpClusterInfo(); // JW enhanced logging
int rc = striper->stat(fr.name, (uint64_t*)&(buf.st_size), &(buf.st_atime)); //Get details about a file
bool fileExists = (rc != -ENOENT); //Make clear what condition we are testing
logwrapper((char*)"Access Mode: %s flags&O_ACCMODE %d ", pathname, flags);
if ((flags & O_ACCMODE) == O_RDONLY) { // Access mode is READ
if (fileExists) {
librados::bufferlist d_stripeUnit;
librados::bufferlist d_objectSize;
std::string obj_name;
librados::IoCtx *context = getIoCtx(fr);
// read first stripe of the object for xattr stripe unit and object size
// this will fail if the object was not written in stripes e.g. s3
// TBD: fallback to direct object (no stripe id appends to filename,
// replace striper metadata with corresponding metadata)
//
try {
obj_name = fr.name + std::string(".0000000000000000");
} catch (std::bad_alloc&) {
logwrapper((char*)"Can not create object string for file %s)", fr.name.c_str());
return -ENOMEM;
}
int ret = 0;
ret = context->getxattr(obj_name, "striper.layout.stripe_unit", d_stripeUnit);
ret = std::min(ret,context->getxattr(obj_name, "striper.layout.object_size", d_objectSize));
//log_func((char*)"size xattr for %s , %llu ,%llu", file_ref->name.c_str(), file_ref->objectSize, file_ref->stripeUnit );
if (ret<=0){
logwrapper((char*)"Could not find size or stripe_unit xattr for %s", fr.name.c_str());
}
else{
//librados's c_str() method does not return a NULL-terminated string, hence why we need to cleanup here
char cleanStripeUnit[MAXDIGITSIZE];
char cleanObjectSize[MAXDIGITSIZE];
unsigned int stripeUnitLength = std::min((unsigned int)MAXDIGITSIZE-1, d_stripeUnit.length());
unsigned int objectSizeLength = std::min((unsigned int)MAXDIGITSIZE-1, d_objectSize.length());
(void)strncpy( cleanStripeUnit, d_stripeUnit.c_str(), stripeUnitLength );
(void)strncpy( cleanObjectSize, d_objectSize.c_str(), objectSizeLength );
cleanStripeUnit[stripeUnitLength] = '\0';
cleanObjectSize[objectSizeLength] = '\0';
//only change defaults if different
if(fr.stripeUnit != std::stoull(cleanStripeUnit)){
logwrapper((char*)"WARNING: stripe unit of %s does not match defaults. object size is %s", pathname, cleanStripeUnit);
fr.stripeUnit = std::stoull(cleanStripeUnit);
}
if(fr.objectSize != std::stoull(cleanObjectSize)){
logwrapper((char*)"WARNING: object size of %s does not match defaults. object size is %s",pathname, cleanObjectSize);
fr.objectSize = std::stoull(cleanObjectSize);
}
}
int fd = insertFileRef(fr);
logwrapper((char*)"File descriptor %d associated to file %s opened in read mode", fd, pathname);
return fd;
} else {
return -ENOENT;
}
} else { // Access mode is WRITE
if (fileExists) {
if (flags & O_TRUNC) {
int rc = ceph_posix_unlink(env, pathname);
if (rc < 0 && rc != -ENOENT) {
return rc;
}
} else {
if (flags & O_EXCL) {
return -EACCES; // permission denied
} else {
return -EEXIST; // otherwise return just file exists
}
}
}
// At this point, we know either the target file didn't exist, or the ceph_posix_unlink above removed it
int fd = insertFileRef(fr);
logwrapper((char*)"File descriptor %d associated to file %s opened in write mode", fd, pathname);
return fd;
}
}
int ceph_posix_close(int fd) {
CephFileRef* fr = getFileRef(fd);
if (fr) {
::timeval now;
::gettimeofday(&now, nullptr);
XrdSysMutexHelper lock(fr->statsMutex);
double lastAsyncAge = 0.0;
// Only compute an age if the starting point was set.
if (fr->lastAsyncSubmission.tv_sec && fr->lastAsyncSubmission.tv_usec) {
lastAsyncAge = 1.0 * (now.tv_sec - fr->lastAsyncSubmission.tv_sec)
+ 0.000001 * (now.tv_usec - fr->lastAsyncSubmission.tv_usec);
}
logwrapper((char*)"ceph_close: closed fd %d for file %s, read ops count %d, write ops count %d, "
"async write ops %d/%d, async pending write bytes %ld, "
"async read ops %d/%d, bytes written/max offset %ld/%ld, "
"longest async write %f, longest callback invocation %f, last async op age %f",
fd, fr->name.c_str(), fr->rdcount, fr->wrcount,
fr->asyncWrCompletionCount, fr->asyncWrStartCount, fr->bytesAsyncWritePending,
fr->asyncRdCompletionCount, fr->asyncRdStartCount, fr->bytesWritten, fr->maxOffsetWritten,
fr->longestAsyncWriteTime, fr->longestCallbackInvocation, (lastAsyncAge));
deleteFileRef(fd, *fr);
return 0;
} else {
return -EBADF;
}
}
static off64_t lseek_compute_offset(CephFileRef &fr, off64_t offset, int whence) {
switch (whence) {
case SEEK_SET:
fr.offset = offset;
break;
case SEEK_CUR:
fr.offset += offset;
break;
default:
return -EINVAL;
}
return fr.offset;
}
off_t ceph_posix_lseek(int fd, off_t offset, int whence) {
CephFileRef* fr = getFileRef(fd);
if (fr) {
logwrapper((char*)"ceph_lseek: for fd %d, offset=%lld, whence=%d", fd, offset, whence);
return (off_t)lseek_compute_offset(*fr, offset, whence);
} else {
return -EBADF;
}
}
off64_t ceph_posix_lseek64(int fd, off64_t offset, int whence) {
CephFileRef* fr = getFileRef(fd);
if (fr) {
logwrapper((char*)"ceph_lseek64: for fd %d, offset=%lld, whence=%d", fd, offset, whence);
return lseek_compute_offset(*fr, offset, whence);
} else {
return -EBADF;
}
}
ssize_t ceph_posix_write(int fd, const void *buf, size_t count) {
CephFileRef* fr = getFileRef(fd);
if (fr) {
logwrapper((char*)"ceph_write: for fd %d, count=%d", fd, count);
if ((fr->flags & O_ACCMODE) == O_RDONLY) {
return -EBADF;
}
libradosstriper::RadosStriper *striper = getRadosStriper(*fr);
if (0 == striper) {
return -EINVAL;
}
ceph::bufferlist bl;
bl.append((const char*)buf, count);
int rc = striper->write(fr->name, bl, count, fr->offset);
if (rc) return rc;
fr->offset += count;
XrdSysMutexHelper lock(fr->statsMutex);
fr->wrcount++;
fr->bytesWritten+=count;
if (fr->offset) fr->maxOffsetWritten = std::max(fr->offset - 1, fr->maxOffsetWritten);
return count;
} else {
return -EBADF;
}
}
ssize_t ceph_posix_pwrite(int fd, const void *buf, size_t count, off64_t offset) {
CephFileRef* fr = getFileRef(fd);
if (fr) {
// TODO implement proper logging level for this plugin - this should be only debug
//logwrapper((char*)"ceph_write: for fd %d, count=%d", fd, count);
if ((fr->flags & O_ACCMODE) == O_RDONLY) {
return -EBADF;
}
libradosstriper::RadosStriper *striper = getRadosStriper(*fr);
if (0 == striper) {
return -EINVAL;
}
ceph::bufferlist bl;
bl.append((const char*)buf, count);
int rc = striper->write(fr->name, bl, count, offset);
if (rc) return rc;
XrdSysMutexHelper lock(fr->statsMutex);
fr->wrcount++;
fr->bytesWritten+=count;
if (offset + count) fr->maxOffsetWritten = std::max(uint64_t(offset + count - 1), fr->maxOffsetWritten);
return count;
} else {
return -EBADF;
}
}
static void ceph_aio_write_complete(rados_completion_t c, void *arg) {
AioArgs *awa = reinterpret_cast<AioArgs*>(arg);
size_t rc = rados_aio_get_return_value(c);
// Compute statistics before reportng to xrootd, so that a close cannot happen
// in the meantime.
CephFileRef* fr = getFileRef(awa->fd);
if (fr) {
XrdSysMutexHelper lock(fr->statsMutex);
fr->asyncWrCompletionCount++;
fr->bytesAsyncWritePending -= awa->nbBytes;
fr->bytesWritten += awa->nbBytes;
if (awa->aiop->sfsAio.aio_nbytes)
fr->maxOffsetWritten = std::max(fr->maxOffsetWritten, uint64_t(awa->aiop->sfsAio.aio_offset + awa->aiop->sfsAio.aio_nbytes - 1));
::timeval now;
::gettimeofday(&now, nullptr);
double writeTime = 0.000001 * (now.tv_usec - awa->startTime.tv_usec) + 1.0 * (now.tv_sec - awa->startTime.tv_sec);
fr->longestAsyncWriteTime = std::max(fr->longestAsyncWriteTime, writeTime);
}
::timeval before, after;
if (fr) ::gettimeofday(&before, nullptr);
awa->callback(awa->aiop, rc == 0 ? awa->nbBytes : rc);
if (fr) {
::gettimeofday(&after, nullptr);
double callbackInvocationTime = 0.000001 * (after.tv_usec - before.tv_usec) + 1.0 * (after.tv_sec - before.tv_sec);
XrdSysMutexHelper lock(fr->statsMutex);
fr->longestCallbackInvocation = std::max(fr->longestCallbackInvocation, callbackInvocationTime);
}
delete(awa);
}
ssize_t ceph_aio_write(int fd, XrdSfsAio *aiop, AioCB *cb) {
CephFileRef* fr = getFileRef(fd);
if (fr) {
// get the parameters from the Xroot aio object
size_t count = aiop->sfsAio.aio_nbytes;
const char *buf = (const char*)aiop->sfsAio.aio_buf;
size_t offset = aiop->sfsAio.aio_offset;
// TODO implement proper logging level for this plugin - this should be only debug
//logwrapper((char*)"ceph_aio_write: for fd %d, count=%d", fd, count);
if ((fr->flags & O_ACCMODE) == O_RDONLY) {
return -EBADF;
}
// get the striper object
libradosstriper::RadosStriper *striper = getRadosStriper(*fr);
if (0 == striper) {
return -EINVAL;
}
// prepare a bufferlist around the given buffer
ceph::bufferlist bl;
bl.append(buf, count);
// get the poolIdx to use
int cephPoolIdx = getCephPoolIdxAndIncrease();
// Get the cluster to use
librados::Rados* cluster = checkAndCreateCluster(cephPoolIdx);
if (0 == cluster) {
return -EINVAL;
}
// prepare a ceph AioCompletion object and do async call
AioArgs *args = new AioArgs(aiop, cb, count, fd);
librados::AioCompletion *completion =
cluster->aio_create_completion(args, ceph_aio_write_complete, NULL);
// do the write
int rc = striper->aio_write(fr->name, completion, bl, count, offset);
completion->release();
XrdSysMutexHelper lock(fr->statsMutex);
fr->asyncWrStartCount++;
::gettimeofday(&fr->lastAsyncSubmission, nullptr);
fr->bytesAsyncWritePending+=count;
return rc;
} else {
return -EBADF;
}
}
ssize_t ceph_nonstriper_readv(int fd, XrdOucIOVec *readV, int n) {
CephFileRef* fr = getFileRef(fd);
if (fr) {
// TODO implement proper logging level for this plugin - this should be only debug
//logwrapper((char*)"ceph_read: for fd %d, count=%d", fd, count);
if ((fr->flags & O_ACCMODE) == O_WRONLY) {
return -EBADF;
}
if (fr->nbStripes != 1) {
//Non-striper based read method works only with a single stripe
return -ENOTSUP;
}
ssize_t read_bytes;
int rc;
librados::IoCtx *ioctx = getIoCtx(*fr);
if (0 == ioctx) {
return -EINVAL;
}
try {
//Constructor can throw bad alloc
bulkAioRead readOp(ioctx, logwrapper, fr);
for (int i = 0; i < n; i++) {
rc = readOp.read(readV[i].data, readV[i].size, readV[i].offset);
if (rc < 0) {
logwrapper( (char*)"Can not declare read request\n");
return rc;
}
}
std::time_t wait_time = std::time(0);
rc = readOp.submit_and_wait_for_complete();
wait_time = std::time(0) - wait_time;
if (wait_time > g_cephAioWaitThresh) {
logwrapper(
(char*)"Waiting for AIO results in readv for %s took %ld seconds, too long!\n",
fr->name.c_str(),
wait_time
);
}
if (rc < 0) {
logwrapper( (char*)"Can not submit read requests\n");
return rc;
}
read_bytes = readOp.get_results();
XrdSysMutexHelper lock(fr->statsMutex);
//We consider readv as a single operation
fr->rdcount += 1;
return read_bytes;
} catch(std::bad_alloc&) {
return -ENOMEM;
}
} else {
return -EBADF;
}
}
ssize_t ceph_striper_readv(int fd, XrdOucIOVec *readV, int n) {
ssize_t nbytes = 0, curCount = 0;
for (int i=0; i<n; i++) {
curCount = ceph_posix_pread(fd, (void *)readV[i].data, (size_t)readV[i].size, (off_t)readV[i].offset);
if (curCount != readV[i].size) {
if (curCount < 0) return curCount;
return -ESPIPE;
}
nbytes += curCount;
}
return nbytes;
}
ssize_t ceph_posix_read(int fd, void *buf, size_t count) {
CephFileRef* fr = getFileRef(fd);
if (fr) {
// TODO implement proper logging level for this plugin - this should be only debug
//logwrapper((char*)"ceph_read: for fd %d, count=%d", fd, count);
if ((fr->flags & O_ACCMODE) == O_WRONLY) {
return -EBADF;
}
libradosstriper::RadosStriper *striper = getRadosStriper(*fr);
if (0 == striper) {
return -EINVAL;
}
ceph::bufferlist bl;
int rc = striper->read(fr->name, &bl, count, fr->offset);
if (rc < 0) return rc;
bl.begin().copy(rc, (char*)buf);
XrdSysMutexHelper lock(fr->statsMutex);
fr->offset += rc;
fr->rdcount++;
return rc;
} else {
return -EBADF;
}
}
ssize_t ceph_posix_nonstriper_pread(int fd, void *buf, size_t count, off64_t offset) {
//The same as pread, but do not relies on rados striper library. Uses direct atomic
//reads from ceph object (see BulkAioRead class for details).
CephFileRef* fr = getFileRef(fd);
if (fr) {
// TODO implement proper logging level for this plugin - this should be only debug
//logwrapper((char*)"ceph_read: for fd %d, count=%d", fd, count);
if ((fr->flags & O_ACCMODE) == O_WRONLY) {
return -EBADF;
}
if (fr->nbStripes != 1) {
//Non-striper based read method works only with a single stripe
return -ENOTSUP;
}
int rc;
ssize_t bytes_read;
librados::IoCtx *ioctx = getIoCtx(*fr);
if (0 == ioctx) {
return -EINVAL;
}
try {
//Constructor can throw bad alloc
bulkAioRead readOp(ioctx, logwrapper, fr);
rc = readOp.read(buf, count, offset);
if (rc < 0) {
logwrapper( (char*)"Can not declare read request\n");
return rc;
}
std::time_t wait_time = std::time(0);
rc = readOp.submit_and_wait_for_complete();
wait_time = std::time(0) - wait_time;
if (wait_time > g_cephAioWaitThresh) {
logwrapper(
(char*)"Waiting for AIO results in pread for %s took %ld seconds, too long!\n",
fr->name.c_str(),
wait_time
);
}
if (rc < 0) {
logwrapper( (char*)"Can not submit read request\n");
return rc;
}
bytes_read = readOp.get_results();
if (bytes_read > 0) {
XrdSysMutexHelper lock(fr->statsMutex);
fr->rdcount++;
} else {
logwrapper( (char*)"Error while read: %d\n", bytes_read);
}
return bytes_read;
} catch (std::bad_alloc&) {
return -ENOMEM;
}
} else {
return -EBADF;
}
}
ssize_t ceph_posix_pread(int fd, void *buf, size_t count, off64_t offset) {
CephFileRef* fr = getFileRef(fd);
if (fr) {
// TODO implement proper logging level for this plugin - this should be only debug
//logwrapper((char*)"ceph_read: for fd %d, count=%d", fd, count);
if ((fr->flags & O_ACCMODE) == O_WRONLY) {
return -EBADF;
}
libradosstriper::RadosStriper *striper = getRadosStriper(*fr);
if (0 == striper) {
return -EINVAL;
}
ceph::bufferlist bl;
int rc = striper->read(fr->name, &bl, count, offset);
if (rc < 0) return rc;
bl.begin().copy(rc, (char*)buf);
XrdSysMutexHelper lock(fr->statsMutex);
fr->rdcount++;
return rc;
} else {
return -EBADF;
}
}
ssize_t ceph_posix_maybestriper_pread(int fd, void *buf, size_t count, off64_t offset, bool allowStriper) {
ssize_t rc {0};
if (!allowStriper) {
rc = ceph_posix_pread(fd,buf,count,offset);
return rc;
}
rc = ceph_posix_nonstriper_pread(fd, buf, count,offset);
if (-ENOENT == rc || -ENOTSUP == rc) {
//This might be a sparse file or nbstripes > 1, so let's try striper read
rc = ceph_posix_pread(fd, buf, count,offset);
if (rc >= 0) {
char err_str[100]; //99 symbols should be enough for the short message
snprintf(err_str, 100, "WARNING! The file (fd %d) seem to be sparse, this is not expected", fd);
logwrapper(err_str);
}
}
return rc;
}
static void ceph_aio_read_complete(rados_completion_t c, void *arg) {
AioArgs *awa = reinterpret_cast<AioArgs*>(arg);
size_t rc = rados_aio_get_return_value(c);
if (awa->bl) {
if (rc > 0) {
awa->bl->begin().copy(rc, (char*)awa->aiop->sfsAio.aio_buf);
}
delete awa->bl;
awa->bl = 0;
}
// Compute statistics before reportng to xrootd, so that a close cannot happen
// in the meantime.
CephFileRef* fr = getFileRef(awa->fd);
if (fr) {
XrdSysMutexHelper lock(fr->statsMutex);
fr->asyncRdCompletionCount++;
}
awa->callback(awa->aiop, rc );
delete(awa);
}
ssize_t ceph_aio_read(int fd, XrdSfsAio *aiop, AioCB *cb) {
CephFileRef* fr = getFileRef(fd);
if (fr) {
// get the parameters from the Xroot aio object
size_t count = aiop->sfsAio.aio_nbytes;
size_t offset = aiop->sfsAio.aio_offset;
// TODO implement proper logging level for this plugin - this should be only debug
//logwrapper((char*)"ceph_aio_read: for fd %d, count=%d", fd, count);
if ((fr->flags & O_ACCMODE) == O_WRONLY) {
return -EBADF;
}
// get the striper object
libradosstriper::RadosStriper *striper = getRadosStriper(*fr);
if (0 == striper) {
return -EINVAL;
}
// prepare a bufferlist to receive data
ceph::bufferlist *bl = new ceph::bufferlist();
// get the poolIdx to use
int cephPoolIdx = getCephPoolIdxAndIncrease();
// Get the cluster to use
librados::Rados* cluster = checkAndCreateCluster(cephPoolIdx);
if (0 == cluster) {
return -EINVAL;
}
// prepare a ceph AioCompletion object and do async call
AioArgs *args = new AioArgs(aiop, cb, count, fd, bl);
librados::AioCompletion *completion =
cluster->aio_create_completion(args, ceph_aio_read_complete, NULL);
// do the read
int rc = striper->aio_read(fr->name, completion, bl, count, offset);
completion->release();
XrdSysMutexHelper lock(fr->statsMutex);
fr->asyncRdStartCount++;
return rc;
} else {
return -EBADF;
}
}
int ceph_posix_fstat(int fd, struct stat *buf) {
CephFileRef* fr = getFileRef(fd);
if (fr) {
logwrapper((char*)__FUNCTION__,": fd %d", fd);
// minimal stat : only size and times are filled
// atime, mtime and ctime are set all to the same value
// mode is set arbitrarily to 0666 | S_IFREG
libradosstriper::RadosStriper *striper = getRadosStriper(*fr);
if (0 == striper) {
logwrapper((char*)"ceph_stat: getRadosStriper failed");
return -EINVAL;
}
memset(buf, 0, sizeof(*buf));
int rc = striper->stat(fr->name, (uint64_t*)&(buf->st_size), &(buf->st_atime));
if (rc != 0) {
return -rc;
}
buf->st_dev = 1;
buf->st_ino = 1;
buf->st_mtime = buf->st_atime;
buf->st_ctime = buf->st_atime;
buf->st_mode = 0666 | S_IFREG;
return 0;
} else {
return -EBADF;
}
}
int ceph_posix_stat(XrdOucEnv* env, const char *pathname, struct stat *buf) {
logwrapper((char*)__FUNCTION__, pathname);
// minimal stat : only size and times are filled
// atime, mtime and ctime are set all to the same value
// mode is set arbitrarily to 0666 | S_IFREG
CephFile file = getCephFile(pathname, env);
libradosstriper::RadosStriper *striper = getRadosStriper(file);
if (0 == striper) {
return -EINVAL;
}
memset(buf, 0, sizeof(*buf));
int rc = striper->stat(file.name, (uint64_t*)&(buf->st_size), &(buf->st_atime));
if (rc != 0) {
// for non existing file. Check that we did not open it for write recently
// in that case, we return 0 size and current time
if (-ENOENT == rc && isOpenForWrite(file.name)) {
buf->st_size = 0;
buf->st_atime = time(NULL);
} else {
return -rc;
}
}
// XRootD assumes an 'offline' file if st_dev and st_ino
// are zero. Set to non-zero (meaningful) values to avoid this
buf->st_dev = 1;
buf->st_ino = 1;
buf->st_mtime = buf->st_atime;
buf->st_ctime = buf->st_atime;
buf->st_mode = 0666 | S_IFREG;
return 0;
}
int ceph_posix_fsync(int fd) {
CephFileRef* fr = getFileRef(fd);
if (fr) {
// no locking of fr as it is not used.
logwrapper((char*)"ceph_sync: fd %d", fd);
return 0;
} else {
return -EBADF;
}
}
int ceph_posix_fcntl(int fd, int cmd, ... /* arg */ ) {
CephFileRef* fr = getFileRef(fd);
if (fr) {
logwrapper((char*)"ceph_fcntl: fd %d cmd=%d", fd, cmd);
// minimal implementation
switch (cmd) {
case F_GETFL:
return fr->mode;
default:
return -EINVAL;
}
} else {
return -EBADF;
}
}
static ssize_t ceph_posix_internal_getxattr(const CephFile &file, const char* name,
void* value, size_t size) {
libradosstriper::RadosStriper *striper = getRadosStriper(file);
if (0 == striper) {
return -EINVAL;
}
ceph::bufferlist bl;
int rc = striper->getxattr(file.name, name, bl);
if (rc < 0) return rc;
size_t returned_size = (size_t)rc<size?rc:size;
bl.begin().copy(returned_size, (char*)value);
return returned_size;
}
ssize_t ceph_posix_getxattr(XrdOucEnv* env, const char* path,
const char* name, void* value,
size_t size) {
logwrapper((char*)"ceph_getxattr: path %s name=%s", path, name);
return ceph_posix_internal_getxattr(getCephFile(path, env), name, value, size);
}
ssize_t ceph_posix_fgetxattr(int fd, const char* name,
void* value, size_t size) {
CephFileRef* fr = getFileRef(fd);
if (fr) {
logwrapper((char*)"ceph_fgetxattr: fd %d name=%s", fd, name);
return ceph_posix_internal_getxattr(*fr, name, value, size);
} else {
return -EBADF;
}
}
static ssize_t ceph_posix_internal_setxattr(const CephFile &file, const char* name,
const void* value, size_t size, int flags) {
libradosstriper::RadosStriper *striper = getRadosStriper(file);
if (0 == striper) {
return -EINVAL;
}
ceph::bufferlist bl;
bl.append((const char*)value, size);
int rc = striper->setxattr(file.name, name, bl);
if (rc) {
return -rc;
}
return 0;
}
ssize_t ceph_posix_setxattr(XrdOucEnv* env, const char* path,
const char* name, const void* value,
size_t size, int flags) {
logwrapper((char*)"ceph_setxattr: path %s name=%s value=%s", path, name, value);
return ceph_posix_internal_setxattr(getCephFile(path, env), name, value, size, flags);
}
int ceph_posix_fsetxattr(int fd,
const char* name, const void* value,
size_t size, int flags) {
CephFileRef* fr = getFileRef(fd);
if (fr) {
logwrapper((char*)"ceph_fsetxattr: fd %d name=%s value=%s", fd, name, value);
return ceph_posix_internal_setxattr(*fr, name, value, size, flags);
} else {
return -EBADF;
}
}
static int ceph_posix_internal_removexattr(const CephFile &file, const char* name) {
libradosstriper::RadosStriper *striper = getRadosStriper(file);
if (0 == striper) {
return -EINVAL;
}
int rc = striper->rmxattr(file.name, name);
if (rc) {
return -rc;
}
return 0;
}
int ceph_posix_removexattr(XrdOucEnv* env, const char* path,
const char* name) {
logwrapper((char*)"ceph_removexattr: path %s name=%s", path, name);
return ceph_posix_internal_removexattr(getCephFile(path, env), name);
}
int ceph_posix_fremovexattr(int fd, const char* name) {
CephFileRef* fr = getFileRef(fd);
if (fr) {
logwrapper((char*)"ceph_fremovexattr: fd %d name=%s", fd, name);
return ceph_posix_internal_removexattr(*fr, name);
} else {
return -EBADF;
}
}
static int ceph_posix_internal_listxattrs(const CephFile &file, XrdSysXAttr::AList **aPL, int getSz) {
libradosstriper::RadosStriper *striper = getRadosStriper(file);
if (0 == striper) {
return -EINVAL;
}
// call ceph
std::map<std::string, ceph::bufferlist> attrset;
int rc = striper->getxattrs(file.name, attrset);
if (rc) {
return -rc;
}
// build result
*aPL = 0;
int maxSize = 0;
for (std::map<std::string, ceph::bufferlist>::const_iterator it = attrset.begin();
it != attrset.end();
it++) {
XrdSysXAttr::AList* newItem = (XrdSysXAttr::AList*)malloc(sizeof(XrdSysXAttr::AList)+it->first.size());
newItem->Next = *aPL;
newItem->Vlen = it->second.length();
if (newItem->Vlen > maxSize) {
maxSize = newItem->Vlen;
}
newItem->Nlen = it->first.size();
strncpy(newItem->Name, it->first.c_str(), newItem->Vlen+1);
*aPL = newItem;
}
if (getSz) {
return 0;
} else {
return maxSize;
}
}
int ceph_posix_listxattrs(XrdOucEnv* env, const char* path, XrdSysXAttr::AList **aPL, int getSz) {
logwrapper((char*)"ceph_listxattrs: path %s", path);
return ceph_posix_internal_listxattrs(getCephFile(path, env), aPL, getSz);
}
int ceph_posix_flistxattrs(int fd, XrdSysXAttr::AList **aPL, int getSz) {
CephFileRef* fr = getFileRef(fd);
if (fr) {
logwrapper((char*)"ceph_flistxattrs: fd %d", fd);
return ceph_posix_internal_listxattrs(*fr, aPL, getSz);
} else {
return -EBADF;
}
}
void ceph_posix_freexattrlist(XrdSysXAttr::AList *aPL) {
while (aPL) {
free(aPL->Name);
XrdSysXAttr::AList *cur = aPL;
aPL = aPL->Next;
free(cur);
}
}
int ceph_posix_statfs(long long *totalSpace, long long *freeSpace) {
logwrapper((char*)"ceph_posix_statfs");
// get the poolIdx to use
int cephPoolIdx = getCephPoolIdxAndIncrease();
// Get the cluster to use
librados::Rados* cluster = checkAndCreateCluster(cephPoolIdx);
if (0 == cluster) {
return -EINVAL;
}
// call ceph stat
librados::cluster_stat_t result;
int rc = cluster->cluster_stat(result);
if (0 == rc) {
*totalSpace = result.kb * 1024;
*freeSpace = result.kb_avail * 1024;
}
return rc;
}
int ceph_posix_stat_pool(char const *poolName, long long *usedSpace) {
logwrapper((char*)__FUNCTION__, poolName);
// get the poolIdx to use
int cephPoolIdx = getCephPoolIdxAndIncrease();
librados::Rados* cluster = checkAndCreateCluster(cephPoolIdx);
if (0 == cluster) {
return -EINVAL;
}
std::list<std::string> poolNames({poolName});
std::map<std::string, librados::pool_stat_t> stat;
if (cluster->get_pool_stats(poolNames, stat) < 0) {
logwrapper((char*)"Unable to get_pool_stats for pool ", poolName);
return -EINVAL;
} else {
*usedSpace = stat[poolName].num_kb * 1024;
return XrdOssOK;
}
}
static int ceph_posix_internal_truncate(const CephFile &file, unsigned long long size) {
libradosstriper::RadosStriper *striper = getRadosStriper(file);
if (0 == striper) {
return -EINVAL;
}
return striper->trunc(file.name, size);
}
int ceph_posix_ftruncate(int fd, unsigned long long size) {
CephFileRef* fr = getFileRef(fd);
if (fr) {
logwrapper((char*)"ceph_posix_ftruncate: fd %d, size %d", fd, size);
return ceph_posix_internal_truncate(*fr, size);
} else {
return -EBADF;
}
}
int ceph_posix_truncate(XrdOucEnv* env, const char *pathname, unsigned long long size) {
logwrapper((char*)"ceph_posix_truncate : %s", pathname);
// minimal stat : only size and times are filled
CephFile file = getCephFile(pathname, env);
return ceph_posix_internal_truncate(file, size);
}
int ceph_posix_unlink(XrdOucEnv* env, const char *pathname) {
logwrapper((char*)"ceph_posix_unlink : %s", pathname);
// start the timer
auto timer_start = std::chrono::steady_clock::now();
// minimal stat : only size and times are filled
CephFile file = getCephFile(pathname, env);
libradosstriper::RadosStriper *striper = getRadosStriper(file);
if (0 == striper) {
return -EINVAL;
}
int rc = striper->remove(file.name);
auto end = std::chrono::steady_clock::now();
auto deltime_ms = std::chrono::duration_cast<std::chrono::milliseconds>(end - timer_start).count();
if (rc == 0) {
logwrapper((char*)"ceph_posix_unlink : %s unlink successful: %d ms", pathname, deltime_ms);
return 0;
}
if (rc != -EBUSY) {
logwrapper((char*)"ceph_posix_unlink : %s unlink failed: %d ms; return code %d", pathname, deltime_ms, rc);
return rc;
}
// if EBUSY returned, assume the file is locked; so try to remove the lock
logwrapper((char*)"ceph_posix_unlink : unlink failed with -EBUSY %s, now trying to remove lock.", pathname);
// lock name is only exposed in the libradosstriper source file, so hardcode it here.
rc = ceph_posix_internal_removexattr(file, "lock.striper.lock");
if (rc !=0 ) {
logwrapper((char*)"ceph_posix_unlink : unlink rmxattr failed %s, %d", pathname, rc);
return rc;
}
// now try to remove again
rc = striper->remove(file.name);
end = std::chrono::steady_clock::now();
deltime_ms = std::chrono::duration_cast<std::chrono::milliseconds>(end - timer_start).count();
if (rc != 0) {
logwrapper((char*)"ceph_posix_unlink : unlink failed after lock removal %s, %d ms", pathname, deltime_ms);
} else {
logwrapper((char*)"ceph_posix_unlink : unlink succeeded after lock removal %s, %d ms", pathname, deltime_ms);
}
return rc;
}
DIR* ceph_posix_opendir(XrdOucEnv* env, const char *pathname) {
logwrapper((char*)"ceph_posix_opendir : %s", pathname);
// only accept root dir, as there is no concept of dirs in object stores
CephFile file = getCephFile(pathname, env);
if (file.name.size() != 1 || file.name[0] != '/') {
errno = -ENOENT;
return 0;
}
librados::IoCtx *ioctx = getIoCtx(file);
if (0 == ioctx) {
errno = EINVAL;
return 0;
}
DirIterator* res = new DirIterator();
res->m_iterator = ioctx->nobjects_begin();
res->m_ioctx = ioctx;
return (DIR*)res;
}
int ceph_posix_readdir(DIR *dirp, char *buff, int blen) {
librados::NObjectIterator &iterator = ((DirIterator*)dirp)->m_iterator;
librados::IoCtx *ioctx = ((DirIterator*)dirp)->m_ioctx;
while (iterator->get_oid().compare(iterator->get_oid().size()-17, 17, ".0000000000000000") &&
iterator != ioctx->nobjects_end()) {
iterator++;
}
if (iterator == ioctx->nobjects_end()) {
buff[0] = 0;
} else {
int l = iterator->get_oid().size()-17;
if (l < blen) blen = l;
strncpy(buff, iterator->get_oid().c_str(), blen-1);
buff[blen-1] = 0;
iterator++;
}
return 0;
}
int ceph_posix_closedir(DIR *dirp) {
delete ((DirIterator*)dirp);
return 0;
}
XrdCephBulkAioRead.hh
g_namelib
XrdOucName2Name * g_namelib
Definition XrdCephPosix.cc:98
g_maxCephPoolIdx
unsigned int g_maxCephPoolIdx
Definition XrdCephPosix.cc:95
logwrapper
static void logwrapper(char *format, va_list argp)
Definition XrdCephOss.cc:62
g_cephAioWaitThresh
unsigned int g_cephAioWaitThresh
Definition XrdCephPosix.cc:91
g_cephPoolIdx
unsigned int g_cephPoolIdx
index of current Striper/IoCtx to be used
Definition XrdCephPosix.cc:88
getRadosStriper
static libradosstriper::RadosStriper * getRadosStriper(const CephFile &file)
Definition XrdCephPosix.cc:597
ceph_posix_maybestriper_pread
ssize_t ceph_posix_maybestriper_pread(int fd, void *buf, size_t count, off64_t offset, bool allowStriper)
Definition XrdCephPosix.cc:1134
ceph_posix_write
ssize_t ceph_posix_write(int fd, const void *buf, size_t count)
Definition XrdCephPosix.cc:821
getFileRef
CephFileRef * getFileRef(int fd)
look for a FileRef from its file descriptor
Definition XrdCephPosix.cc:149
getCephPoolIdxAndIncrease
unsigned int getCephPoolIdxAndIncrease()
Definition XrdCephPosix.cc:117
ceph_posix_set_logfunc
void ceph_posix_set_logfunc(void(*logfunc)(char *, va_list argp))
Definition XrdCephPosix.cc:644
ceph_posix_truncate
int ceph_posix_truncate(XrdOucEnv *env, const char *pathname, unsigned long long size)
Definition XrdCephPosix.cc:1535
g_defaultParams
CephFile g_defaultParams
global variable containing defaults for CephFiles
Definition XrdCephPosix.cc:190
ceph_aio_write
ssize_t ceph_aio_write(int fd, XrdSfsAio *aiop, AioCB *cb)
Definition XrdCephPosix.cc:903
ceph_posix_unlink
int ceph_posix_unlink(XrdOucEnv *env, const char *pathname)
Definition XrdCephPosix.cc:1542
fillCephFileParams
void fillCephFileParams(const std::string &params, XrdOucEnv *env, CephFile &file)
Definition XrdCephPosix.cc:385
ceph_aio_read
ssize_t ceph_aio_read(int fd, XrdSfsAio *aiop, AioCB *cb)
Definition XrdCephPosix.cc:1175
ceph_posix_readdir
int ceph_posix_readdir(DIR *dirp, char *buff, int blen)
Definition XrdCephPosix.cc:1607
g_filesOpenForWrite
std::multiset< std::string > g_filesOpenForWrite
global variable holding a list of files currently opened for write
Definition XrdCephPosix.cc:101
ceph_posix_fcntl
int ceph_posix_fcntl(int fd, int cmd,...)
Definition XrdCephPosix.cc:1286
ceph_posix_internal_setxattr
static ssize_t ceph_posix_internal_setxattr(const CephFile &file, const char *name, const void *value, size_t size, int flags)
Definition XrdCephPosix.cc:1334
ceph_posix_getxattr
ssize_t ceph_posix_getxattr(XrdOucEnv *env, const char *path, const char *name, void *value, size_t size)
Definition XrdCephPosix.cc:1316
g_cluster
std::vector< librados::Rados * > g_cluster
Definition XrdCephPosix.cc:84
fillCephStripeUnit
static int fillCephStripeUnit(const std::string &params, unsigned int offset, XrdOucEnv *env, CephFile &file)
Definition XrdCephPosix.cc:341
ceph_aio_write_complete
static void ceph_aio_write_complete(rados_completion_t c, void *arg)
Definition XrdCephPosix.cc:873
getCephFile
static CephFile getCephFile(const char *path, XrdOucEnv *env)
Definition XrdCephPosix.cc:447
ceph_nonstriper_readv
ssize_t ceph_nonstriper_readv(int fd, XrdOucIOVec *readV, int n)
Definition XrdCephPosix.cc:947
g_striper_mutex
XrdSysMutex g_striper_mutex
mutex protecting the striper and ioctx maps
Definition XrdCephPosix.cc:86
StriperDict
std::map< std::string, libradosstriper::RadosStriper * > StriperDict
Definition XrdCephPosix.cc:80
ceph_posix_read
ssize_t ceph_posix_read(int fd, void *buf, size_t count)
Definition XrdCephPosix.cc:1023
fillCephPool
static int fillCephPool(const std::string &params, unsigned int offset, XrdOucEnv *env, CephFile &file)
Definition XrdCephPosix.cc:289
ceph_posix_pread
ssize_t ceph_posix_pread(int fd, void *buf, size_t count, off64_t offset)
Definition XrdCephPosix.cc:1110
ceph_posix_listxattrs
int ceph_posix_listxattrs(XrdOucEnv *env, const char *path, XrdSysXAttr::AList **aPL, int getSz)
Definition XrdCephPosix.cc:1430
ceph_posix_fstat
int ceph_posix_fstat(int fd, struct stat *buf)
Definition XrdCephPosix.cc:1215
stoui
static unsigned int stoui(const std::string &s)
simple integer parsing, to be replaced by std::stoi when C++11 can be used
Definition XrdCephPosix.cc:226
ceph_posix_disconnect_all
void ceph_posix_disconnect_all()
Definition XrdCephPosix.cc:624
ceph_posix_fsync
int ceph_posix_fsync(int fd)
Definition XrdCephPosix.cc:1275
g_init_mutex
XrdSysMutex g_init_mutex
mutex protecting initialization of ceph clusters
Definition XrdCephPosix.cc:109
ceph_posix_closedir
int ceph_posix_closedir(DIR *dirp)
Definition XrdCephPosix.cc:1626
fillCephNbStripes
static int fillCephNbStripes(const std::string &params, unsigned int offset, XrdOucEnv *env, CephFile &file)
Definition XrdCephPosix.cc:315
g_fds
std::map< unsigned int, CephFileRef > g_fds
global variable holding a map of file descriptor to file reference
Definition XrdCephPosix.cc:103
getIoCtx
static librados::IoCtx * getIoCtx(const CephFile &file)
Definition XrdCephPosix.cc:611
g_nextCephFd
unsigned int g_nextCephFd
global variable remembering the next available file descriptor
Definition XrdCephPosix.cc:105
ceph_posix_internal_getxattr
static ssize_t ceph_posix_internal_getxattr(const CephFile &file, const char *name, void *value, size_t size)
Definition XrdCephPosix.cc:1302
g_ioCtx
std::vector< IOCtxDict > g_ioCtx
Definition XrdCephPosix.cc:83
insertFileRef
int insertFileRef(CephFileRef &fr)
Definition XrdCephPosix.cc:179
ceph_posix_opendir
DIR * ceph_posix_opendir(XrdOucEnv *env, const char *pathname)
Definition XrdCephPosix.cc:1588
ceph_posix_statfs
int ceph_posix_statfs(long long *totalSpace, long long *freeSpace)
Definition XrdCephPosix.cc:1454
checkAndCreateStriper
int checkAndCreateStriper(unsigned int cephPoolIdx, std::string &userAtPool, const CephFile &file)
Definition XrdCephPosix.cc:510
fillCephObjectSize
static void fillCephObjectSize(const std::string &params, unsigned int offset, XrdOucEnv *env, CephFile &file)
Definition XrdCephPosix.cc:367
getCephFileRef
static CephFileRef getCephFileRef(const char *path, XrdOucEnv *env, int flags, mode_t mode, unsigned long long offset)
Definition XrdCephPosix.cc:453
ceph_posix_fremovexattr
int ceph_posix_fremovexattr(int fd, const char *name)
Definition XrdCephPosix.cc:1386
checkAndCreateCluster
librados::Rados * checkAndCreateCluster(unsigned int cephPoolIdx, std::string userId=g_defaultParams.userId)
Definition XrdCephPosix.cc:476
translateFileName
void translateFileName(std::string &physName, std::string logName)
converts a logical filename to physical one if needed
Definition XrdCephPosix.cc:406
ceph_posix_close
int ceph_posix_close(int fd)
Definition XrdCephPosix.cc:760
ceph_posix_stat_pool
int ceph_posix_stat_pool(char const *poolName, long long *usedSpace)
Return the amount of space used in a pool.
Definition XrdCephPosix.cc:1491
DirIterator::m_ioctx
librados::IoCtx * m_ioctx
Definition XrdCephPosix.cc:62
g_idxCntr
std::map< unsigned int, unsigned long long > g_idxCntr
Definition XrdCephPosix.cc:112
IOCtxDict
std::map< std::string, librados::IoCtx * > IOCtxDict
Definition XrdCephPosix.cc:82
ceph_posix_lseek
off_t ceph_posix_lseek(int fd, off_t offset, int whence)
Definition XrdCephPosix.cc:801
ceph_posix_set_defaults
void ceph_posix_set_defaults(const char *value)
Definition XrdCephPosix.cc:397
deleteFileRef
void deleteFileRef(int fd, const CephFileRef &fr)
deletes a FileRef from the global table of file descriptors
Definition XrdCephPosix.cc:164
ceph_posix_fsetxattr
int ceph_posix_fsetxattr(int fd, const char *name, const void *value, size_t size, int flags)
Definition XrdCephPosix.cc:1356
ceph_aio_read_complete
static void ceph_aio_read_complete(rados_completion_t c, void *arg)
Definition XrdCephPosix.cc:1154
ceph_posix_ftruncate
int ceph_posix_ftruncate(int fd, unsigned long long size)
Definition XrdCephPosix.cc:1525
dumpClusterInfo
void dumpClusterInfo()
Definition XrdCephPosix.cc:241
ceph_posix_internal_removexattr
static int ceph_posix_internal_removexattr(const CephFile &file, const char *name)
Definition XrdCephPosix.cc:1368
g_radosStripers
std::vector< StriperDict > g_radosStripers
Definition XrdCephPosix.cc:81
ceph_posix_open
int ceph_posix_open(XrdOucEnv *env, const char *pathname, int flags, mode_t mode)
Definition XrdCephPosix.cc:664
DirIterator::m_iterator
librados::NObjectIterator m_iterator
Definition XrdCephPosix.cc:61
ceph_posix_nonstriper_pread
ssize_t ceph_posix_nonstriper_pread(int fd, void *buf, size_t count, off64_t offset)
Definition XrdCephPosix.cc:1048
stoull
static unsigned long long int stoull(const std::string &s)
simple integer parsing, to be replaced by std::stoll when C++11 can be used
Definition XrdCephPosix.cc:212
logwrapper
static void logwrapper(char *format,...)
Definition XrdCephPosix.cc:203
ceph_posix_removexattr
int ceph_posix_removexattr(XrdOucEnv *env, const char *path, const char *name)
Definition XrdCephPosix.cc:1380
ceph_posix_lseek64
off64_t ceph_posix_lseek64(int fd, off64_t offset, int whence)
Definition XrdCephPosix.cc:811
ceph_striper_readv
ssize_t ceph_striper_readv(int fd, XrdOucIOVec *readV, int n)
Definition XrdCephPosix.cc:1007
ceph_posix_stat
int ceph_posix_stat(XrdOucEnv *env, const char *pathname, struct stat *buf)
Definition XrdCephPosix.cc:1243
isOpenForWrite
bool isOpenForWrite(std::string &name)
check whether a file is open for write
Definition XrdCephPosix.cc:143
fillCephUserId
static int fillCephUserId(const std::string &params, XrdOucEnv *env, CephFile &file)
Definition XrdCephPosix.cc:268
fillCephFile
void fillCephFile(const char *path, XrdOucEnv *env, CephFile &file)
fill a ceph file struct from a path and an environment
Definition XrdCephPosix.cc:424
ceph_posix_pwrite
ssize_t ceph_posix_pwrite(int fd, const void *buf, size_t count, off64_t offset)
Definition XrdCephPosix.cc:847
g_defaultPool
std::string g_defaultPool
Definition XrdCephPosix.cc:198
g_fd_mutex
XrdSysMutex g_fd_mutex
mutex protecting the map of file descriptors and the openForWrite multiset
Definition XrdCephPosix.cc:107
ceph_posix_setxattr
ssize_t ceph_posix_setxattr(XrdOucEnv *env, const char *path, const char *name, const void *value, size_t size, int flags)
Definition XrdCephPosix.cc:1349
ceph_posix_internal_truncate
static int ceph_posix_internal_truncate(const CephFile &file, unsigned long long size)
Definition XrdCephPosix.cc:1517
g_logfunc
static void(* g_logfunc)(char *, va_list argp)=0
global variable for the log function
Definition XrdCephPosix.cc:201
g_defaultUserId
std::string g_defaultUserId
Definition XrdCephPosix.cc:197
ceph_posix_flistxattrs
int ceph_posix_flistxattrs(int fd, XrdSysXAttr::AList **aPL, int getSz)
Definition XrdCephPosix.cc:1435
ceph_posix_fgetxattr
ssize_t ceph_posix_fgetxattr(int fd, const char *name, void *value, size_t size)
Definition XrdCephPosix.cc:1323
ceph_posix_internal_listxattrs
static int ceph_posix_internal_listxattrs(const CephFile &file, XrdSysXAttr::AList **aPL, int getSz)
Definition XrdCephPosix.cc:1396
lseek_compute_offset
static off64_t lseek_compute_offset(CephFileRef &fr, off64_t offset, int whence)
Definition XrdCephPosix.cc:787
ceph_posix_freexattrlist
void ceph_posix_freexattrlist(XrdSysXAttr::AList *aPL)
Definition XrdCephPosix.cc:1445
DirIterator
small struct for directory listing
Definition XrdCephPosix.cc:60
XrdCephPosix.hh
AioCB
void AioCB(XrdSfsAio *, size_t)
Definition XrdCephPosix.hh:58
MAXDIGITSIZE
#define MAXDIGITSIZE
Definition XrdCephPosix.hh:45
XrdOss.hh
XrdOssOK
#define XrdOssOK
Definition XrdOss.hh:50
XrdOucIOVec.hh
XrdOucName2Name.hh
stat
#define stat(a, b)
Definition XrdPosix.hh:101
XrdSfsAio.hh
aiocb::aio_offset
off_t aio_offset
Definition XrdSfsAio.hh:49
aiocb::aio_nbytes
size_t aio_nbytes
Definition XrdSfsAio.hh:48
aiocb::aio_buf
void * aio_buf
Definition XrdSfsAio.hh:47
XrdSfsFlags.hh
XrdSysPlatform.hh
XrdSysPthread.hh
XrdOucEnv
Definition XrdOucEnv.hh:42
XrdOucEnv::Get
char * Get(const char *varname)
Definition XrdOucEnv.hh:69
XrdOucName2Name
Definition XrdOucName2Name.hh:49
XrdSfsAio
Definition XrdSfsAio.hh:59
XrdSfsAio::sfsAio
struct aiocb sfsAio
Definition XrdSfsAio.hh:62
XrdSysMutexHelper
Definition XrdSysPthread.hh:263
XrdSysMutex
Definition XrdSysPthread.hh:165
XrdSysXAttr::AList::Name
char Name[1]
Start of the name (size of struct is dynamic).
Definition XrdSysXAttr.hh:56
XrdSysXAttr::AList::Vlen
int Vlen
The length of the attribute value;.
Definition XrdSysXAttr.hh:54
XrdSysXAttr::AList::Nlen
int Nlen
The length of the attribute name that follows.
Definition XrdSysXAttr.hh:55
XrdSysXAttr::AList::Next
AList * Next
-> next element.
Definition XrdSysXAttr.hh:53
XrdSysXAttr::AList
Definition XrdSysXAttr.hh:53
bulkAioRead
Definition XrdCephBulkAioRead.hh:10
bulkAioRead::get_results
ssize_t get_results()
Definition XrdCephBulkAioRead.cc:114
bulkAioRead::read
int read(void *out_buf, size_t size, off64_t offset)
Definition XrdCephBulkAioRead.cc:141
bulkAioRead::submit_and_wait_for_complete
int submit_and_wait_for_complete()
Definition XrdCephBulkAioRead.cc:64
AioArgs
small struct for aio API callbacks
Definition XrdCephPosix.cc:66
AioArgs::nbBytes
size_t nbBytes
Definition XrdCephPosix.cc:71
AioArgs::startTime
::timeval startTime
Definition XrdCephPosix.cc:73
AioArgs::fd
int fd
Definition XrdCephPosix.cc:72
AioArgs::callback
AioCB * callback
Definition XrdCephPosix.cc:70
AioArgs::AioArgs
AioArgs(XrdSfsAio *a, AioCB *b, size_t n, int _fd, ceph::bufferlist *_bl=0)
Definition XrdCephPosix.cc:67
AioArgs::aiop
XrdSfsAio * aiop
Definition XrdCephPosix.cc:69
AioArgs::bl
ceph::bufferlist * bl
Definition XrdCephPosix.cc:74
CephFileRef
Definition XrdCephPosix.hh:112
CephFileRef::bytesAsyncWritePending
uint64_t bytesAsyncWritePending
Definition XrdCephPosix.hh:119
CephFileRef::asyncRdStartCount
unsigned asyncRdStartCount
Definition XrdCephPosix.hh:123
CephFileRef::asyncWrStartCount
unsigned asyncWrStartCount
Definition XrdCephPosix.hh:125
CephFileRef::maxOffsetWritten
uint64_t maxOffsetWritten
Definition XrdCephPosix.hh:118
CephFileRef::lastAsyncSubmission
::timeval lastAsyncSubmission
Definition XrdCephPosix.hh:127
CephFileRef::longestCallbackInvocation
double longestCallbackInvocation
Definition XrdCephPosix.hh:129
CephFileRef::mode
mode_t mode
Definition XrdCephPosix.hh:114
CephFileRef::bytesWritten
uint64_t bytesWritten
Definition XrdCephPosix.hh:120
CephFileRef::offset
uint64_t offset
Definition XrdCephPosix.hh:115
CephFileRef::longestAsyncWriteTime
double longestAsyncWriteTime
Definition XrdCephPosix.hh:128
CephFileRef::wrcount
unsigned wrcount
Definition XrdCephPosix.hh:122
CephFileRef::rdcount
unsigned rdcount
Definition XrdCephPosix.hh:121
CephFileRef::statsMutex
XrdSysMutex statsMutex
Definition XrdCephPosix.hh:117
CephFileRef::asyncWrCompletionCount
unsigned asyncWrCompletionCount
Definition XrdCephPosix.hh:126
CephFileRef::flags
int flags
Definition XrdCephPosix.hh:113
CephFileRef::asyncRdCompletionCount
unsigned asyncRdCompletionCount
Definition XrdCephPosix.hh:124
CephFile
small structs to store file metadata
Definition XrdCephPosix.hh:103
CephFile::userId
std::string userId
Definition XrdCephPosix.hh:106
CephFile::nbStripes
unsigned int nbStripes
Definition XrdCephPosix.hh:107
CephFile::pool
std::string pool
Definition XrdCephPosix.hh:105
CephFile::stripeUnit
unsigned long long stripeUnit
Definition XrdCephPosix.hh:108
CephFile::name
std::string name
Definition XrdCephPosix.hh:104
CephFile::objectSize
unsigned long long objectSize
Definition XrdCephPosix.hh:109
XrdOucIOVec
Definition XrdOucIOVec.hh:41