XrdCephPosix.cc File Reference

#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"

Include dependency graph for XrdCephPosix.cc:

Go to the source code of this file.

Classes
struct	AioArgs
	small struct for aio API callbacks More...
struct	DirIterator
	small struct for directory listing More...

Typedefs
typedef std::map< std::string, librados::IoCtx * >	IOCtxDict
typedef std::map< std::string, libradosstriper::RadosStriper * >	StriperDict

Functions
ssize_t	ceph_aio_read (int fd, XrdSfsAio *aiop, AioCB *cb)
static void	ceph_aio_read_complete (rados_completion_t c, void *arg)
ssize_t	ceph_aio_write (int fd, XrdSfsAio *aiop, AioCB *cb)
static void	ceph_aio_write_complete (rados_completion_t c, void *arg)
ssize_t	ceph_nonstriper_readv (int fd, XrdOucIOVec *readV, int n)
int	ceph_posix_close (int fd)
int	ceph_posix_closedir (DIR *dirp)
void	ceph_posix_disconnect_all ()
int	ceph_posix_fcntl (int fd, int cmd,...)
ssize_t	ceph_posix_fgetxattr (int fd, const char *name, void *value, size_t size)
int	ceph_posix_flistxattrs (int fd, XrdSysXAttr::AList **aPL, int getSz)
void	ceph_posix_freexattrlist (XrdSysXAttr::AList *aPL)
int	ceph_posix_fremovexattr (int fd, const char *name)
int	ceph_posix_fsetxattr (int fd, const char *name, const void *value, size_t size, int flags)
int	ceph_posix_fstat (int fd, struct stat *buf)
int	ceph_posix_fsync (int fd)
int	ceph_posix_ftruncate (int fd, unsigned long long size)
ssize_t	ceph_posix_getxattr (XrdOucEnv *env, const char *path, const char *name, void *value, size_t size)
static ssize_t	ceph_posix_internal_getxattr (const CephFile &file, const char *name, void *value, size_t size)
static int	ceph_posix_internal_listxattrs (const CephFile &file, XrdSysXAttr::AList **aPL, int getSz)
static int	ceph_posix_internal_removexattr (const CephFile &file, const char *name)
static ssize_t	ceph_posix_internal_setxattr (const CephFile &file, const char *name, const void *value, size_t size, int flags)
static int	ceph_posix_internal_truncate (const CephFile &file, unsigned long long size)
int	ceph_posix_listxattrs (XrdOucEnv *env, const char *path, XrdSysXAttr::AList **aPL, int getSz)
off_t	ceph_posix_lseek (int fd, off_t offset, int whence)
off64_t	ceph_posix_lseek64 (int fd, off64_t offset, int whence)
ssize_t	ceph_posix_maybestriper_pread (int fd, void *buf, size_t count, off64_t offset, bool allowStriper)
ssize_t	ceph_posix_nonstriper_pread (int fd, void *buf, size_t count, off64_t offset)
int	ceph_posix_open (XrdOucEnv *env, const char *pathname, int flags, mode_t mode)
DIR *	ceph_posix_opendir (XrdOucEnv *env, const char *pathname)
ssize_t	ceph_posix_pread (int fd, void *buf, size_t count, off64_t offset)
ssize_t	ceph_posix_pwrite (int fd, const void *buf, size_t count, off64_t offset)
ssize_t	ceph_posix_read (int fd, void *buf, size_t count)
int	ceph_posix_readdir (DIR *dirp, char *buff, int blen)
int	ceph_posix_removexattr (XrdOucEnv *env, const char *path, const char *name)
void	ceph_posix_set_defaults (const char *value)
void	ceph_posix_set_logfunc (void(*logfunc)(char *, va_list argp))
ssize_t	ceph_posix_setxattr (XrdOucEnv *env, const char *path, const char *name, const void *value, size_t size, int flags)
int	ceph_posix_stat (XrdOucEnv *env, const char *pathname, struct stat *buf)
int	ceph_posix_stat_pool (char const *poolName, long long *usedSpace)
	Return the amount of space used in a pool.
int	ceph_posix_statfs (long long *totalSpace, long long *freeSpace)
int	ceph_posix_truncate (XrdOucEnv *env, const char *pathname, unsigned long long size)
int	ceph_posix_unlink (XrdOucEnv *env, const char *pathname)
ssize_t	ceph_posix_write (int fd, const void *buf, size_t count)
ssize_t	ceph_striper_readv (int fd, XrdOucIOVec *readV, int n)
librados::Rados *	checkAndCreateCluster (unsigned int cephPoolIdx, std::string userId=g_defaultParams.userId)
int	checkAndCreateStriper (unsigned int cephPoolIdx, std::string &userAtPool, const CephFile &file)
void	deleteFileRef (int fd, const CephFileRef &fr)
	deletes a FileRef from the global table of file descriptors
void	dumpClusterInfo ()
void	fillCephFile (const char *path, XrdOucEnv *env, CephFile &file)
	fill a ceph file struct from a path and an environment
void	fillCephFileParams (const std::string &params, XrdOucEnv *env, CephFile &file)
static int	fillCephNbStripes (const std::string &params, unsigned int offset, XrdOucEnv *env, CephFile &file)
static void	fillCephObjectSize (const std::string &params, unsigned int offset, XrdOucEnv *env, CephFile &file)
static int	fillCephPool (const std::string &params, unsigned int offset, XrdOucEnv *env, CephFile &file)
static int	fillCephStripeUnit (const std::string &params, unsigned int offset, XrdOucEnv *env, CephFile &file)
static int	fillCephUserId (const std::string &params, XrdOucEnv *env, CephFile &file)
static CephFile	getCephFile (const char *path, XrdOucEnv *env)
static CephFileRef	getCephFileRef (const char *path, XrdOucEnv *env, int flags, mode_t mode, unsigned long long offset)
unsigned int	getCephPoolIdxAndIncrease ()
CephFileRef *	getFileRef (int fd)
	look for a FileRef from its file descriptor
static librados::IoCtx *	getIoCtx (const CephFile &file)
static libradosstriper::RadosStriper *	getRadosStriper (const CephFile &file)
int	insertFileRef (CephFileRef &fr)
bool	isOpenForWrite (std::string &name)
	check whether a file is open for write
static void	logwrapper (char *format,...)
static off64_t	lseek_compute_offset (CephFileRef &fr, off64_t offset, int whence)
static unsigned int	stoui (const std::string &s)
	simple integer parsing, to be replaced by std::stoi when C++11 can be used
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
void	translateFileName (std::string &physName, std::string logName)
	converts a logical filename to physical one if needed

Variables
unsigned int	g_cephAioWaitThresh = 15
unsigned int	g_cephPoolIdx = 0
	index of current Striper/IoCtx to be used
std::vector< librados::Rados * >	g_cluster
CephFile	g_defaultParams
	global variable containing defaults for CephFiles
std::string	g_defaultPool = "default"
std::string	g_defaultUserId = "admin"
XrdSysMutex	g_fd_mutex
	mutex protecting the map of file descriptors and the openForWrite multiset
std::map< unsigned int, CephFileRef >	g_fds
	global variable holding a map of file descriptor to file reference
std::multiset< std::string >	g_filesOpenForWrite
	global variable holding a list of files currently opened for write
std::map< unsigned int, unsigned long long >	g_idxCntr
XrdSysMutex	g_init_mutex
	mutex protecting initialization of ceph clusters
std::vector< IOCtxDict >	g_ioCtx
static void(*	g_logfunc )(char *, va_list argp)=0
	global variable for the log function
unsigned int	g_maxCephPoolIdx = 1
XrdOucName2Name *	g_namelib = 0
unsigned int	g_nextCephFd = 0
	global variable remembering the next available file descriptor
std::vector< StriperDict >	g_radosStripers
XrdSysMutex	g_striper_mutex
	mutex protecting the striper and ioctx maps

---

Class Documentation

DirIterator

struct DirIterator

small struct for directory listing

Definition at line 60 of file XrdCephPosix.cc.

Collaboration diagram for DirIterator:

Class Members
IoCtx *	m_ioctx
NObjectIterator	m_iterator

Typedef Documentation

IOCtxDict

typedef std::map<std::string, librados::IoCtx*> IOCtxDict

Definition at line 82 of file XrdCephPosix.cc.

StriperDict

typedef std::map<std::string, libradosstriper::RadosStriper*> StriperDict

global variables holding stripers/ioCtxs/cluster objects Note that we have a pool of them to circumvent the limitation of having a single objecter/messenger per IoCtx

Definition at line 80 of file XrdCephPosix.cc.

Function Documentation

ceph_aio_read()

ssize_t ceph_aio_read	(	int	fd,
		XrdSfsAio *	aiop,
		AioCB *	cb )

Definition at line 1175 of file XrdCephPosix.cc.

                                                          {
  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;
  }
}

References aiocb::aio_nbytes, aiocb::aio_offset, CephFileRef::asyncRdStartCount, ceph_aio_read_complete(), checkAndCreateCluster(), CephFileRef::flags, getCephPoolIdxAndIncrease(), getFileRef(), getRadosStriper(), CephFile::name, XrdSfsAio::sfsAio, and CephFileRef::statsMutex.

Referenced by XrdCephOssFile::Read(), and XrdCephBuffer::CephIOAdapterAIORaw::read().

Here is the call graph for this function:

Here is the caller graph for this function:

ceph_aio_read_complete()

static void ceph_aio_read_complete ( rados_completion_t c, void * arg )

static

Definition at line 1154 of file XrdCephPosix.cc.

                                                                    {
  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);
}

References aiocb::aio_buf, AioArgs::aiop, CephFileRef::asyncRdCompletionCount, AioArgs::bl, AioArgs::callback, AioArgs::fd, getFileRef(), XrdSfsAio::sfsAio, and CephFileRef::statsMutex.

Referenced by ceph_aio_read().

Here is the call graph for this function:

Here is the caller graph for this function:

ceph_aio_write()

ssize_t ceph_aio_write	(	int	fd,
		XrdSfsAio *	aiop,
		AioCB *	cb )

Definition at line 903 of file XrdCephPosix.cc.

                                                           {
  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;
  }
}

References aiocb::aio_buf, aiocb::aio_nbytes, aiocb::aio_offset, CephFileRef::asyncWrStartCount, CephFileRef::bytesAsyncWritePending, ceph_aio_write_complete(), checkAndCreateCluster(), CephFileRef::flags, getCephPoolIdxAndIncrease(), getFileRef(), getRadosStriper(), CephFileRef::lastAsyncSubmission, CephFile::name, XrdSfsAio::sfsAio, and CephFileRef::statsMutex.

Referenced by XrdCephOssFile::Write(), and XrdCephBuffer::CephIOAdapterAIORaw::write().

Here is the call graph for this function:

Here is the caller graph for this function:

ceph_aio_write_complete()

static void ceph_aio_write_complete ( rados_completion_t c, void * arg )

static

Definition at line 873 of file XrdCephPosix.cc.

                                                                     {
  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);
}

References aiocb::aio_nbytes, aiocb::aio_offset, AioArgs::aiop, CephFileRef::asyncWrCompletionCount, CephFileRef::bytesAsyncWritePending, CephFileRef::bytesWritten, AioArgs::callback, AioArgs::fd, getFileRef(), CephFileRef::longestAsyncWriteTime, CephFileRef::longestCallbackInvocation, CephFileRef::maxOffsetWritten, AioArgs::nbBytes, XrdSfsAio::sfsAio, AioArgs::startTime, and CephFileRef::statsMutex.

Referenced by ceph_aio_write().

Here is the call graph for this function:

Here is the caller graph for this function:

ceph_nonstriper_readv()

ssize_t ceph_nonstriper_readv	(	int	fd,
		XrdOucIOVec *	readV,
		int	n )

Definition at line 947 of file XrdCephPosix.cc.

                                                                 {
  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;
  }
}

References CephFileRef::flags, g_cephAioWaitThresh, bulkAioRead::get_results(), getFileRef(), getIoCtx(), logwrapper(), logwrapper(), CephFile::name, CephFile::nbStripes, CephFileRef::rdcount, bulkAioRead::read(), CephFileRef::statsMutex, and bulkAioRead::submit_and_wait_for_complete().

Referenced by XrdCephOssFile::ReadV().

Here is the call graph for this function:

Here is the caller graph for this function:

ceph_posix_close()

int ceph_posix_close

(

int

fd

)

Definition at line 760 of file XrdCephPosix.cc.

                             {
  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;
  }
}

References CephFileRef::asyncRdCompletionCount, CephFileRef::asyncRdStartCount, CephFileRef::asyncWrCompletionCount, CephFileRef::asyncWrStartCount, CephFileRef::bytesAsyncWritePending, CephFileRef::bytesWritten, deleteFileRef(), getFileRef(), CephFileRef::lastAsyncSubmission, logwrapper(), CephFileRef::longestAsyncWriteTime, CephFileRef::longestCallbackInvocation, CephFileRef::maxOffsetWritten, CephFile::name, CephFileRef::rdcount, CephFileRef::statsMutex, and CephFileRef::wrcount.

Referenced by XrdCephOssFile::Close().

Here is the call graph for this function:

Here is the caller graph for this function:

ceph_posix_closedir()

int ceph_posix_closedir

(

DIR *

dirp

)

Definition at line 1626 of file XrdCephPosix.cc.

                                   {
  delete ((DirIterator*)dirp);
  return 0;
}

Referenced by XrdCephOssDir::Close().

Here is the caller graph for this function:

ceph_posix_disconnect_all()

void ceph_posix_disconnect_all

(

)

Definition at line 624 of file XrdCephPosix.cc.

                                 {
  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();
}

References g_cluster, g_ioCtx, g_maxCephPoolIdx, g_radosStripers, and g_striper_mutex.

Referenced by XrdCephOss::~XrdCephOss().

Here is the caller graph for this function:

ceph_posix_fcntl()

int ceph_posix_fcntl	(	int	fd,
		int	cmd,
			... )

Definition at line 1286 of file XrdCephPosix.cc.

                                              {
  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;
  }
}

References getFileRef(), logwrapper(), and CephFileRef::mode.

Here is the call graph for this function:

ceph_posix_fgetxattr()

ssize_t ceph_posix_fgetxattr	(	int	fd,
		const char *	name,
		void *	value,
		size_t	size )

Definition at line 1323 of file XrdCephPosix.cc.

                                                       {
  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;
  }
}

References ceph_posix_internal_getxattr(), getFileRef(), and logwrapper().

Referenced by XrdCephXAttr::Get().

Here is the call graph for this function:

Here is the caller graph for this function:

ceph_posix_flistxattrs()

int ceph_posix_flistxattrs	(	int	fd,
		XrdSysXAttr::AList **	aPL,
		int	getSz )

Definition at line 1435 of file XrdCephPosix.cc.

                                                                      {
  CephFileRef* fr = getFileRef(fd);
  if (fr) {
    logwrapper((char*)"ceph_flistxattrs: fd %d", fd);
    return ceph_posix_internal_listxattrs(*fr, aPL, getSz);
  } else {
    return -EBADF;
  }
}

References ceph_posix_internal_listxattrs(), getFileRef(), and logwrapper().

Referenced by XrdCephXAttr::List().

Here is the call graph for this function:

Here is the caller graph for this function:

ceph_posix_freexattrlist()

void ceph_posix_freexattrlist

(

XrdSysXAttr::AList *

aPL

)

Definition at line 1445 of file XrdCephPosix.cc.

                                                     {
  while (aPL) {
    free(aPL->Name);
    XrdSysXAttr::AList *cur = aPL;
    aPL = aPL->Next;
    free(cur);
  }
}

References XrdSysXAttr::AList::Name, and XrdSysXAttr::AList::Next.

Referenced by XrdCephXAttr::Free().

Here is the caller graph for this function:

ceph_posix_fremovexattr()

int ceph_posix_fremovexattr	(	int	fd,
		const char *	name )

Definition at line 1386 of file XrdCephPosix.cc.

                                                      {
  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;
  }
}

References ceph_posix_internal_removexattr(), getFileRef(), and logwrapper().

Here is the call graph for this function:

ceph_posix_fsetxattr()

int ceph_posix_fsetxattr	(	int	fd,
		const char *	name,
		const void *	value,
		size_t	size,
		int	flags )

Definition at line 1356 of file XrdCephPosix.cc.

                                                  {
  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;
  }
}

References ceph_posix_internal_setxattr(), getFileRef(), and logwrapper().

Referenced by XrdCephXAttr::Set().

Here is the call graph for this function:

Here is the caller graph for this function:

ceph_posix_fstat()

int ceph_posix_fstat	(	int	fd,
		struct stat *	buf )

Definition at line 1215 of file XrdCephPosix.cc.

                                               {
  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;
  }
}

References getFileRef(), getRadosStriper(), logwrapper(), CephFile::name, and stat.

Referenced by XrdCephOssFile::Fstat().

Here is the call graph for this function:

Here is the caller graph for this function:

ceph_posix_fsync()

int ceph_posix_fsync

(

int

fd

)

Definition at line 1275 of file XrdCephPosix.cc.

                             {
  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;
  }
}

References getFileRef(), and logwrapper().

Referenced by XrdCephOssFile::Fsync().

Here is the call graph for this function:

Here is the caller graph for this function:

ceph_posix_ftruncate()

int ceph_posix_ftruncate	(	int	fd,
		unsigned long long	size )

Definition at line 1525 of file XrdCephPosix.cc.

                                                          {
  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;
  }
}

References ceph_posix_internal_truncate(), getFileRef(), and logwrapper().

Referenced by XrdCephOssFile::Ftruncate().

Here is the call graph for this function:

Here is the caller graph for this function:

ceph_posix_getxattr()

ssize_t ceph_posix_getxattr	(	XrdOucEnv *	env,
		const char *	path,
		const char *	name,
		void *	value,
		size_t	size )

Definition at line 1316 of file XrdCephPosix.cc.

                                         {
  logwrapper((char*)"ceph_getxattr: path %s name=%s", path, name);
  return ceph_posix_internal_getxattr(getCephFile(path, env), name, value, size);
}

References ceph_posix_internal_getxattr(), getCephFile(), and logwrapper().

Referenced by XrdCephXAttr::Get(), and getNumericAttr().

Here is the call graph for this function:

Here is the caller graph for this function:

ceph_posix_internal_getxattr()

static ssize_t ceph_posix_internal_getxattr ( const CephFile & file, const char * name, void * value, size_t size )

static

Definition at line 1302 of file XrdCephPosix.cc.

                                                                      {
  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;
}

References getRadosStriper(), and CephFile::name.

Referenced by ceph_posix_fgetxattr(), and ceph_posix_getxattr().

Here is the call graph for this function:

Here is the caller graph for this function:

ceph_posix_internal_listxattrs()

static int ceph_posix_internal_listxattrs ( const CephFile & file, XrdSysXAttr::AList ** aPL, int getSz )

static

Definition at line 1396 of file XrdCephPosix.cc.

                                                                                                   {
  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;
  }
}

References getRadosStriper(), XrdSysXAttr::AList::Name, CephFile::name, XrdSysXAttr::AList::Next, XrdSysXAttr::AList::Nlen, and XrdSysXAttr::AList::Vlen.

Referenced by ceph_posix_flistxattrs(), and ceph_posix_listxattrs().

Here is the call graph for this function:

Here is the caller graph for this function:

ceph_posix_internal_removexattr()

static int ceph_posix_internal_removexattr ( const CephFile & file, const char * name )

static

Definition at line 1368 of file XrdCephPosix.cc.

                                                                                   {
  libradosstriper::RadosStriper *striper = getRadosStriper(file);
  if (0 == striper) {
    return -EINVAL;
  }
  int rc = striper->rmxattr(file.name, name);
  if (rc) {
    return -rc;
  }
  return 0;
}

References getRadosStriper(), and CephFile::name.

Referenced by ceph_posix_fremovexattr(), ceph_posix_removexattr(), and ceph_posix_unlink().

Here is the call graph for this function:

Here is the caller graph for this function:

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 )

static

Definition at line 1334 of file XrdCephPosix.cc.

                                                                                       {
  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;
}

References getRadosStriper(), and CephFile::name.

Referenced by ceph_posix_fsetxattr(), and ceph_posix_setxattr().

Here is the call graph for this function:

Here is the caller graph for this function:

ceph_posix_internal_truncate()

static int ceph_posix_internal_truncate ( const CephFile & file, unsigned long long size )

static

Definition at line 1517 of file XrdCephPosix.cc.

                                                                                       {
  libradosstriper::RadosStriper *striper = getRadosStriper(file);
  if (0 == striper) {
    return -EINVAL;
  }
  return striper->trunc(file.name, size);
}

References getRadosStriper(), and CephFile::name.

Referenced by ceph_posix_ftruncate(), and ceph_posix_truncate().

Here is the call graph for this function:

Here is the caller graph for this function:

ceph_posix_listxattrs()

int ceph_posix_listxattrs	(	XrdOucEnv *	env,
		const char *	path,
		XrdSysXAttr::AList **	aPL,
		int	getSz )

Definition at line 1430 of file XrdCephPosix.cc.

                                                                                               {
  logwrapper((char*)"ceph_listxattrs: path %s", path);
  return ceph_posix_internal_listxattrs(getCephFile(path, env), aPL, getSz);
}

References ceph_posix_internal_listxattrs(), getCephFile(), and logwrapper().

Referenced by XrdCephXAttr::List().

Here is the call graph for this function:

Here is the caller graph for this function:

ceph_posix_lseek()

off_t ceph_posix_lseek	(	int	fd,
		off_t	offset,
		int	whence )

Definition at line 801 of file XrdCephPosix.cc.

                                                         {
  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;
  }
}

References getFileRef(), logwrapper(), and lseek_compute_offset().

Here is the call graph for this function:

ceph_posix_lseek64()

off64_t ceph_posix_lseek64	(	int	fd,
		off64_t	offset,
		int	whence )

Definition at line 811 of file XrdCephPosix.cc.

                                                               {
  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;
  }
}

References getFileRef(), logwrapper(), and lseek_compute_offset().

Here is the call graph for this function:

ceph_posix_maybestriper_pread()

ssize_t ceph_posix_maybestriper_pread	(	int	fd,
		void *	buf,
		size_t	count,
		off64_t	offset,
		bool	allowStriper )

Definition at line 1134 of file XrdCephPosix.cc.

                                                                                                          {
  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; 
}

References ceph_posix_nonstriper_pread(), ceph_posix_pread(), and logwrapper().

Referenced by XrdCephBuffer::CephIOAdapterRaw::read(), and XrdCephBuffer::XrdCephBufferAlgSimple::read().

Here is the call graph for this function:

Here is the caller graph for this function:

ceph_posix_nonstriper_pread()

ssize_t ceph_posix_nonstriper_pread	(	int	fd,
		void *	buf,
		size_t	count,
		off64_t	offset )

Definition at line 1048 of file XrdCephPosix.cc.

                                                                                     {
  //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;
  }
}

References CephFileRef::flags, g_cephAioWaitThresh, bulkAioRead::get_results(), getFileRef(), getIoCtx(), logwrapper(), logwrapper(), CephFile::name, CephFile::nbStripes, CephFileRef::rdcount, bulkAioRead::read(), CephFileRef::statsMutex, and bulkAioRead::submit_and_wait_for_complete().

Referenced by ceph_posix_maybestriper_pread(), and XrdCephOssFile::Read().

Here is the call graph for this function:

Here is the caller graph for this function:

ceph_posix_open()

int ceph_posix_open	(	XrdOucEnv *	env,
		const char *	pathname,
		int	flags,
		mode_t	mode )

• brief ceph_posix_open function opens a file for read or write
• details This function either:
• Opens a file for reading. If the file doesn't exist, this is an error.
• Opens a file for writing. If the file already exists, check whether overwrite has been requested. If overwrite
• hasn't been requested for an existing file, this is an error.
• param env XrdOucEnv* Unused
• param pathname const char* Specify the file to open.
• param flags int Indicates whether reading or writing, and whether to overwrite an existing file.
• param mode mode_t Unused
• return int This is a file descriptor (non-negative) if the operation is successful,
• or an error code (negative value) if the operation fails

Definition at line 664 of file XrdCephPosix.cc.

                                                                                 {
 
  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;
    
  }
    
}

References ceph_posix_unlink(), dumpClusterInfo(), getCephFileRef(), getIoCtx(), getRadosStriper(), insertFileRef(), logwrapper(), MAXDIGITSIZE, CephFile::name, CephFile::objectSize, stat, and CephFile::stripeUnit.

Referenced by XrdCephOssFile::Open().

Here is the call graph for this function:

Here is the caller graph for this function:

ceph_posix_opendir()

DIR * ceph_posix_opendir	(	XrdOucEnv *	env,
		const char *	pathname )

Definition at line 1588 of file XrdCephPosix.cc.

                                                              {
  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;
}

References getCephFile(), getIoCtx(), logwrapper(), DirIterator::m_ioctx, DirIterator::m_iterator, and CephFile::name.

Referenced by XrdCephOssDir::Opendir().

Here is the call graph for this function:

Here is the caller graph for this function:

ceph_posix_pread()

ssize_t ceph_posix_pread	(	int	fd,
		void *	buf,
		size_t	count,
		off64_t	offset )

Definition at line 1110 of file XrdCephPosix.cc.

                                                                          {
  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;
  }
}

References CephFileRef::flags, getFileRef(), getRadosStriper(), CephFile::name, CephFileRef::rdcount, and CephFileRef::statsMutex.

Referenced by ceph_posix_maybestriper_pread(), ceph_striper_readv(), and XrdCephOssFile::Read().

Here is the call graph for this function:

Here is the caller graph for this function:

ceph_posix_pwrite()

ssize_t ceph_posix_pwrite	(	int	fd,
		const void *	buf,
		size_t	count,
		off64_t	offset )

Definition at line 847 of file XrdCephPosix.cc.

                                                                                 {
  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;
  }
}

References CephFileRef::bytesWritten, CephFileRef::flags, getFileRef(), getRadosStriper(), CephFileRef::maxOffsetWritten, CephFile::name, CephFileRef::statsMutex, and CephFileRef::wrcount.

Referenced by XrdCephOssFile::Write(), and XrdCephBuffer::CephIOAdapterRaw::write().

Here is the call graph for this function:

Here is the caller graph for this function:

ceph_posix_read()

ssize_t ceph_posix_read	(	int	fd,
		void *	buf,
		size_t	count )

Definition at line 1023 of file XrdCephPosix.cc.

                                                         {
  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;
  }
}

References CephFileRef::flags, getFileRef(), getRadosStriper(), CephFile::name, CephFileRef::offset, CephFileRef::rdcount, and CephFileRef::statsMutex.

Here is the call graph for this function:

ceph_posix_readdir()

int ceph_posix_readdir	(	DIR *	dirp,
		char *	buff,
		int	blen )

Definition at line 1607 of file XrdCephPosix.cc.

                                                        {
  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;
}

References DirIterator::m_iterator.

Referenced by XrdCephOssDir::Readdir().

Here is the caller graph for this function:

ceph_posix_removexattr()

int ceph_posix_removexattr	(	XrdOucEnv *	env,
		const char *	path,
		const char *	name )

Definition at line 1380 of file XrdCephPosix.cc.

                                             {
  logwrapper((char*)"ceph_removexattr: path %s name=%s", path, name);
  return ceph_posix_internal_removexattr(getCephFile(path, env), name);
}

References ceph_posix_internal_removexattr(), getCephFile(), and logwrapper().

Referenced by XrdCephXAttr::Del().

Here is the call graph for this function:

Here is the caller graph for this function:

ceph_posix_set_defaults()

void ceph_posix_set_defaults

(

const char *

value

)

sets the default userId, pool and file layout syntax is [user@]pool[,nbStripes[,stripeUnit[,objectSize]]] may throw std::invalid_argument or std::out_of_range in case of error

Definition at line 397 of file XrdCephPosix.cc.

                                                {
  if (value) {
    CephFile newdefault;
    fillCephFileParams(value, NULL, newdefault);
    g_defaultParams = newdefault;
  }
}

References fillCephFileParams(), and g_defaultParams.

Referenced by XrdOssGetStorageSystem(), and XrdSysGetXAttrObject().

Here is the call graph for this function:

Here is the caller graph for this function:

ceph_posix_set_logfunc()

void ceph_posix_set_logfunc

(

void(*

logfunc )(char *, va_list argp)

)

Definition at line 644 of file XrdCephPosix.cc.

                                                                    {
  g_logfunc = logfunc;
};

References g_logfunc.

Referenced by XrdOssGetStorageSystem().

Here is the caller graph for this function:

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 at line 1349 of file XrdCephPosix.cc.

                                                    {
  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);
}

References ceph_posix_internal_setxattr(), getCephFile(), and logwrapper().

Referenced by XrdCephXAttr::Set().

Here is the call graph for this function:

Here is the caller graph for this function:

ceph_posix_stat()

int ceph_posix_stat	(	XrdOucEnv *	env,
		const char *	pathname,
		struct stat *	buf )

Definition at line 1243 of file XrdCephPosix.cc.

                                                                            {
  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;
}

References getCephFile(), getRadosStriper(), isOpenForWrite(), logwrapper(), CephFile::name, and stat.

Referenced by XrdCephOss::Stat().

Here is the call graph for this function:

Here is the caller graph for this function:

ceph_posix_stat_pool()

int ceph_posix_stat_pool	(	char const *	poolName,
		long long *	usedSpace )

Return the amount of space used in a pool.

This function - Obtains the statistics that librados holds on a pool Calculates the number of bytes allocated to the pool @params poolName: (in) the name of the pool to query usedSpace: (out) the number of bytes used in the pool

Returns

success or failure status

Implementation: Jyothish Thomas STFC RAL, jyoth.nosp@m.ish..nosp@m.thoma.nosp@m.s@st.nosp@m.fc.ac.nosp@m..uk, 2022 Ian Johnson STFC RAL, ian.j.nosp@m.ohns.nosp@m.on@st.nosp@m.fc.a.nosp@m.c.uk, 2022, 2023

Definition at line 1491 of file XrdCephPosix.cc.

                                                                     {
 
  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;
 
  }
}

References checkAndCreateCluster(), getCephPoolIdxAndIncrease(), logwrapper(), stat, and XrdOssOK.

Referenced by XrdCephOss::StatLS().

Here is the call graph for this function:

Here is the caller graph for this function:

ceph_posix_statfs()

int ceph_posix_statfs	(	long long *	totalSpace,
		long long *	freeSpace )

Definition at line 1454 of file XrdCephPosix.cc.

                                                                   {
  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;
}

References checkAndCreateCluster(), getCephPoolIdxAndIncrease(), and logwrapper().

Referenced by XrdCephOss::StatVS().

Here is the call graph for this function:

Here is the caller graph for this function:

ceph_posix_truncate()

int ceph_posix_truncate	(	XrdOucEnv *	env,
		const char *	pathname,
		unsigned long long	size )

Definition at line 1535 of file XrdCephPosix.cc.

                                                                                       {
  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);
}

References ceph_posix_internal_truncate(), getCephFile(), and logwrapper().

Referenced by XrdCephOss::Truncate().

Here is the call graph for this function:

Here is the caller graph for this function:

ceph_posix_unlink()

int ceph_posix_unlink	(	XrdOucEnv *	env,
		const char *	pathname )

Definition at line 1542 of file XrdCephPosix.cc.

                                                            {
  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; 
}

References ceph_posix_internal_removexattr(), getCephFile(), getRadosStriper(), logwrapper(), and CephFile::name.

Referenced by ceph_posix_open(), and XrdCephOss::Unlink().

Here is the call graph for this function:

Here is the caller graph for this function:

ceph_posix_write()

ssize_t ceph_posix_write	(	int	fd,
		const void *	buf,
		size_t	count )

Definition at line 821 of file XrdCephPosix.cc.

                                                                {
  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;
  }
}

References CephFileRef::bytesWritten, CephFileRef::flags, getFileRef(), getRadosStriper(), logwrapper(), CephFileRef::maxOffsetWritten, CephFile::name, CephFileRef::offset, CephFileRef::statsMutex, and CephFileRef::wrcount.

Here is the call graph for this function:

ceph_striper_readv()

ssize_t ceph_striper_readv	(	int	fd,
		XrdOucIOVec *	readV,
		int	n )

Sequential, striper-based readv implementation.

Definition at line 1007 of file XrdCephPosix.cc.

                                                              {
  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;
}

References ceph_posix_pread().

Referenced by XrdCephOssFile::ReadV().

Here is the call graph for this function:

Here is the caller graph for this function:

checkAndCreateCluster()

librados::Rados * checkAndCreateCluster ( unsigned int cephPoolIdx, std::string userId = g_defaultParams.userId )

inline

Definition at line 476 of file XrdCephPosix.cc.

                                                                                         {
  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];
}

References g_cluster, g_defaultParams, and logwrapper().

Referenced by ceph_aio_read(), ceph_aio_write(), ceph_posix_stat_pool(), ceph_posix_statfs(), and checkAndCreateStriper().

Here is the call graph for this function:

Here is the caller graph for this function:

checkAndCreateStriper()

int checkAndCreateStriper	(	unsigned int	cephPoolIdx,
		std::string &	userAtPool,
		const CephFile &	file )

Definition at line 510 of file XrdCephPosix.cc.

                                                                                                 {
  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;
} 

References checkAndCreateCluster(), g_cluster, g_ioCtx, g_radosStripers, logwrapper(), CephFile::nbStripes, CephFile::objectSize, CephFile::pool, CephFile::stripeUnit, and CephFile::userId.

Referenced by getIoCtx(), and getRadosStriper().

Here is the call graph for this function:

Here is the caller graph for this function:

deleteFileRef()

void deleteFileRef	(	int	fd,
		const CephFileRef &	fr )

deletes a FileRef from the global table of file descriptors

Definition at line 164 of file XrdCephPosix.cc.

                                                  {
  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);
  }
}

References CephFileRef::flags, g_fd_mutex, g_fds, g_filesOpenForWrite, and CephFile::name.

Referenced by ceph_posix_close().

Here is the caller graph for this function:

dumpClusterInfo()

void dumpClusterInfo

(

)

Definition at line 241 of file XrdCephPosix.cc.

                       {
  //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());
}

References g_cluster, g_fds, g_filesOpenForWrite, g_idxCntr, g_ioCtx, g_radosStripers, and logwrapper().

Referenced by ceph_posix_open().

Here is the call graph for this function:

Here is the caller graph for this function:

fillCephFile()

void fillCephFile	(	const char *	path,
		XrdOucEnv *	env,
		CephFile &	file )

fill a ceph file struct from a path and an environment

Definition at line 424 of file XrdCephPosix.cc.

                                                                    {
  // 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);
  }
}

References fillCephFileParams(), CephFile::name, and translateFileName().

Referenced by getCephFile(), and getCephFileRef().

Here is the call graph for this function:

Here is the caller graph for this function:

fillCephFileParams()

void fillCephFileParams	(	const std::string &	params,
		XrdOucEnv *	env,
		CephFile &	file )

fill the parameters of a ceph file struct (all but name) from a string and an environment see fillCephFile for the detailed syntax

Definition at line 385 of file XrdCephPosix.cc.

                                                                                 {
  // 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);
}

References fillCephNbStripes(), fillCephObjectSize(), fillCephPool(), fillCephStripeUnit(), and fillCephUserId().

Referenced by ceph_posix_set_defaults(), and fillCephFile().

Here is the call graph for this function:

Here is the caller graph for this function:

fillCephNbStripes()

static int fillCephNbStripes ( const std::string & params, unsigned int offset, XrdOucEnv * env, CephFile & file )

static

fills the nbStriped of a ceph file struct from a string and an environment returns position of first character after the nbStripes

Definition at line 315 of file XrdCephPosix.cc.

                                                                                                           {
  // 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;
  }
}

References g_defaultParams, XrdOucEnv::Get(), CephFile::nbStripes, and stoui().

Referenced by fillCephFileParams().

Here is the call graph for this function:

Here is the caller graph for this function:

fillCephObjectSize()

static void fillCephObjectSize ( const std::string & params, unsigned int offset, XrdOucEnv * env, CephFile & file )

static

fills the objectSize of a ceph file struct from a string and an environment returns position of first character after the objectSize

Definition at line 367 of file XrdCephPosix.cc.

                                                                                                             {
  // 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));
  }
}

References g_defaultParams, XrdOucEnv::Get(), CephFile::objectSize, and stoull().

Referenced by fillCephFileParams().

Here is the call graph for this function:

Here is the caller graph for this function:

fillCephPool()

static int fillCephPool ( const std::string & params, unsigned int offset, XrdOucEnv * env, CephFile & file )

static

fills the pool of a ceph file struct from a string and an environment returns position of first character after the pool

Definition at line 289 of file XrdCephPosix.cc.

                                                                                                      {
  // 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;
  }
}

References g_defaultParams, XrdOucEnv::Get(), and CephFile::pool.

Referenced by fillCephFileParams().

Here is the call graph for this function:

Here is the caller graph for this function:

fillCephStripeUnit()

static int fillCephStripeUnit ( const std::string & params, unsigned int offset, XrdOucEnv * env, CephFile & file )

static

fills the stripeUnit of a ceph file struct from a string and an environment returns position of first character after the stripeUnit

Definition at line 341 of file XrdCephPosix.cc.

                                                                                                            {
  // 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;
  }
}

References g_defaultParams, XrdOucEnv::Get(), stoull(), and CephFile::stripeUnit.

Referenced by fillCephFileParams().

Here is the call graph for this function:

Here is the caller graph for this function:

fillCephUserId()

static int fillCephUserId ( const std::string & params, XrdOucEnv * env, CephFile & file )

static

fills the userId of a ceph file struct from a string and an environment returns position of first character after the userId

Definition at line 268 of file XrdCephPosix.cc.

                                                                                   {
  // 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;
  }
}

References g_defaultParams, XrdOucEnv::Get(), and CephFile::userId.

Referenced by fillCephFileParams().

Here is the call graph for this function:

Here is the caller graph for this function:

getCephFile()

static CephFile getCephFile ( const char * path, XrdOucEnv * env )

static

Definition at line 447 of file XrdCephPosix.cc.

                                                              {
  CephFile file;
  fillCephFile(path, env, file);
  return file;
}

References fillCephFile().

Referenced by ceph_posix_getxattr(), ceph_posix_listxattrs(), ceph_posix_opendir(), ceph_posix_removexattr(), ceph_posix_setxattr(), ceph_posix_stat(), ceph_posix_truncate(), and ceph_posix_unlink().

Here is the call graph for this function:

Here is the caller graph for this function:

getCephFileRef()

static CephFileRef getCephFileRef ( const char * path, XrdOucEnv * env, int flags, mode_t mode, unsigned long long offset )

static

Definition at line 453 of file XrdCephPosix.cc.

                                                                          {
  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;
}

References CephFileRef::asyncRdCompletionCount, CephFileRef::asyncRdStartCount, CephFileRef::asyncWrCompletionCount, CephFileRef::asyncWrStartCount, CephFileRef::bytesAsyncWritePending, CephFileRef::bytesWritten, fillCephFile(), CephFileRef::flags, CephFileRef::lastAsyncSubmission, CephFileRef::longestAsyncWriteTime, CephFileRef::longestCallbackInvocation, CephFileRef::maxOffsetWritten, CephFileRef::mode, CephFileRef::offset, CephFileRef::rdcount, and CephFileRef::wrcount.

Referenced by ceph_posix_open().

Here is the call graph for this function:

Here is the caller graph for this function:

getCephPoolIdxAndIncrease()

unsigned int getCephPoolIdxAndIncrease

(

)

Accessor to next ceph pool index Note that this is not thread safe, but we do not care as we only want a rough load balancing

Definition at line 117 of file XrdCephPosix.cc.

                                         {
  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;
}

References g_cephPoolIdx, g_cluster, g_idxCntr, g_init_mutex, g_ioCtx, g_maxCephPoolIdx, and g_radosStripers.

Referenced by ceph_aio_read(), ceph_aio_write(), ceph_posix_stat_pool(), ceph_posix_statfs(), getIoCtx(), and getRadosStriper().

Here is the caller graph for this function:

getFileRef()

CephFileRef * getFileRef

(

int

fd

)

look for a FileRef from its file descriptor

Definition at line 149 of file XrdCephPosix.cc.

                                {
  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;
  }
}

References g_fd_mutex, and g_fds.

Referenced by ceph_aio_read(), ceph_aio_read_complete(), ceph_aio_write(), ceph_aio_write_complete(), ceph_nonstriper_readv(), ceph_posix_close(), ceph_posix_fcntl(), ceph_posix_fgetxattr(), ceph_posix_flistxattrs(), ceph_posix_fremovexattr(), ceph_posix_fsetxattr(), ceph_posix_fstat(), ceph_posix_fsync(), ceph_posix_ftruncate(), ceph_posix_lseek(), ceph_posix_lseek64(), ceph_posix_nonstriper_pread(), ceph_posix_pread(), ceph_posix_pwrite(), ceph_posix_read(), and ceph_posix_write().

Here is the caller graph for this function:

getIoCtx()

static librados::IoCtx * getIoCtx ( const CephFile & file )

static

Definition at line 611 of file XrdCephPosix.cc.

                                                     {
  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];
}

References checkAndCreateStriper(), g_ioCtx, g_striper_mutex, getCephPoolIdxAndIncrease(), CephFile::nbStripes, CephFile::objectSize, CephFile::pool, CephFile::stripeUnit, and CephFile::userId.

Referenced by ceph_nonstriper_readv(), ceph_posix_nonstriper_pread(), ceph_posix_open(), and ceph_posix_opendir().

Here is the call graph for this function:

Here is the caller graph for this function:

getRadosStriper()

static libradosstriper::RadosStriper * getRadosStriper ( const CephFile & file )

static

Definition at line 597 of file XrdCephPosix.cc.

                                                                          {
  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];
}

References checkAndCreateStriper(), g_radosStripers, g_striper_mutex, getCephPoolIdxAndIncrease(), logwrapper(), CephFile::nbStripes, CephFile::objectSize, CephFile::pool, CephFile::stripeUnit, and CephFile::userId.

Referenced by ceph_aio_read(), ceph_aio_write(), ceph_posix_fstat(), ceph_posix_internal_getxattr(), ceph_posix_internal_listxattrs(), ceph_posix_internal_removexattr(), ceph_posix_internal_setxattr(), ceph_posix_internal_truncate(), ceph_posix_open(), ceph_posix_pread(), ceph_posix_pwrite(), ceph_posix_read(), ceph_posix_stat(), ceph_posix_unlink(), and ceph_posix_write().

Here is the call graph for this function:

Here is the caller graph for this function:

insertFileRef()

int insertFileRef

(

CephFileRef &

fr

)

inserts a new FileRef into the global table of file descriptors and return the associated file descriptor

Definition at line 179 of file XrdCephPosix.cc.

                                   {
  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;
}

References CephFileRef::flags, g_fd_mutex, g_fds, g_filesOpenForWrite, g_nextCephFd, and CephFile::name.

Referenced by ceph_posix_open().

Here is the caller graph for this function:

isOpenForWrite()

bool isOpenForWrite

(

std::string &

name

)

check whether a file is open for write

Definition at line 143 of file XrdCephPosix.cc.

                                     {
  XrdSysMutexHelper lock(g_fd_mutex);
  return g_filesOpenForWrite.find(name) != g_filesOpenForWrite.end();
}

References g_fd_mutex, and g_filesOpenForWrite.

Referenced by ceph_posix_stat().

Here is the caller graph for this function:

logwrapper()

static void logwrapper ( char * format, ... )

static

Definition at line 203 of file XrdCephPosix.cc.

                                          {
  if (0 == g_logfunc) return;
  va_list arg;
  va_start(arg, format);
  (*g_logfunc)(format, arg);
  va_end(arg);
}

References g_logfunc.

Referenced by ceph_nonstriper_readv(), ceph_posix_close(), ceph_posix_fcntl(), ceph_posix_fgetxattr(), ceph_posix_flistxattrs(), ceph_posix_fremovexattr(), ceph_posix_fsetxattr(), ceph_posix_fstat(), ceph_posix_fsync(), ceph_posix_ftruncate(), ceph_posix_getxattr(), ceph_posix_listxattrs(), ceph_posix_lseek(), ceph_posix_lseek64(), ceph_posix_maybestriper_pread(), ceph_posix_nonstriper_pread(), ceph_posix_open(), ceph_posix_opendir(), ceph_posix_removexattr(), ceph_posix_setxattr(), ceph_posix_stat(), ceph_posix_stat_pool(), ceph_posix_statfs(), ceph_posix_truncate(), ceph_posix_unlink(), ceph_posix_write(), checkAndCreateCluster(), checkAndCreateStriper(), dumpClusterInfo(), getRadosStriper(), and translateFileName().

Here is the caller graph for this function:

lseek_compute_offset()

static off64_t lseek_compute_offset ( CephFileRef & fr, off64_t offset, int whence )

static

Definition at line 787 of file XrdCephPosix.cc.

                                                                                 {
  switch (whence) {
  case SEEK_SET:
    fr.offset = offset;
    break;
  case SEEK_CUR:
    fr.offset += offset;
    break;
  default:
    return -EINVAL;
  }
  return fr.offset;
}

References CephFileRef::offset.

Referenced by ceph_posix_lseek(), and ceph_posix_lseek64().

Here is the caller graph for this function:

stoui()

static unsigned int stoui ( const std::string & s )

static

simple integer parsing, to be replaced by std::stoi when C++11 can be used

Definition at line 226 of file XrdCephPosix.cc.

                                              {
  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;
}

Referenced by fillCephNbStripes().

Here is the caller graph for this function:

stoull()

static unsigned long long int stoull ( const std::string & s )

static

simple integer parsing, to be replaced by std::stoll when C++11 can be used

Definition at line 212 of file XrdCephPosix.cc.

                                                         {
  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;
}

Referenced by fillCephObjectSize(), and fillCephStripeUnit().

Here is the caller graph for this function:

translateFileName()

void translateFileName	(	std::string &	physName,
		std::string	logName )

converts a logical filename to physical one if needed

Definition at line 406 of file XrdCephPosix.cc.

                                                              {
  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;
  }
}

References g_namelib, and logwrapper().

Referenced by fillCephFile().

Here is the call graph for this function:

Here is the caller graph for this function:

Variable Documentation

g_cephAioWaitThresh

unsigned int g_cephAioWaitThresh = 15

If aio read operation takes longer than this value, a warning will be issued

Definition at line 91 of file XrdCephPosix.cc.

Referenced by ceph_nonstriper_readv(), ceph_posix_nonstriper_pread(), and XrdCephOss::Configure().

g_cephPoolIdx

unsigned int g_cephPoolIdx = 0

index of current Striper/IoCtx to be used

Definition at line 88 of file XrdCephPosix.cc.

Referenced by getCephPoolIdxAndIncrease().

g_cluster

std::vector<librados::Rados*> g_cluster

Definition at line 84 of file XrdCephPosix.cc.

Referenced by ceph_posix_disconnect_all(), checkAndCreateCluster(), checkAndCreateStriper(), dumpClusterInfo(), and getCephPoolIdxAndIncrease().

g_defaultParams

CephFile g_defaultParams

Initial value:

= { "",
                             "default",        
                             "admin",          
                             1,                
                             4 * 1024 * 1024,  
                             4 * 1024 * 1024}

global variable containing defaults for CephFiles

Definition at line 190 of file XrdCephPosix.cc.

                           { "",
                             "default",        // default pool
                             "admin",          // default user
                             1,                // default nbStripes
                             4 * 1024 * 1024,  // default stripeUnit : 4 MB
                             4 * 1024 * 1024}; // default objectSize : 4 MB

Referenced by ceph_posix_set_defaults(), checkAndCreateCluster(), fillCephNbStripes(), fillCephObjectSize(), fillCephPool(), fillCephStripeUnit(), and fillCephUserId().

g_defaultPool

std::string g_defaultPool = "default"

Definition at line 198 of file XrdCephPosix.cc.

g_defaultUserId

std::string g_defaultUserId = "admin"

Definition at line 197 of file XrdCephPosix.cc.

g_fd_mutex

XrdSysMutex g_fd_mutex

mutex protecting the map of file descriptors and the openForWrite multiset

Definition at line 107 of file XrdCephPosix.cc.

Referenced by deleteFileRef(), getFileRef(), insertFileRef(), and isOpenForWrite().

g_fds

std::map<unsigned int, CephFileRef> g_fds

global variable holding a map of file descriptor to file reference

Definition at line 103 of file XrdCephPosix.cc.

Referenced by deleteFileRef(), dumpClusterInfo(), getFileRef(), and insertFileRef().

g_filesOpenForWrite

std::multiset<std::string> g_filesOpenForWrite

global variable holding a list of files currently opened for write

Definition at line 101 of file XrdCephPosix.cc.

Referenced by deleteFileRef(), dumpClusterInfo(), insertFileRef(), and isOpenForWrite().

g_idxCntr

std::map<unsigned int, unsigned long long> g_idxCntr

Definition at line 112 of file XrdCephPosix.cc.

Referenced by dumpClusterInfo(), and getCephPoolIdxAndIncrease().

g_init_mutex

XrdSysMutex g_init_mutex

mutex protecting initialization of ceph clusters

Definition at line 109 of file XrdCephPosix.cc.

Referenced by getCephPoolIdxAndIncrease().

g_ioCtx

std::vector<IOCtxDict> g_ioCtx

Definition at line 83 of file XrdCephPosix.cc.

Referenced by ceph_posix_disconnect_all(), checkAndCreateStriper(), dumpClusterInfo(), getCephPoolIdxAndIncrease(), and getIoCtx().

g_logfunc

void(* g_logfunc) (char *, va_list argp)=0 ( char * , va_list argp )

static

global variable for the log function

Definition at line 201 of file XrdCephPosix.cc.

Referenced by ceph_posix_set_logfunc(), and logwrapper().

g_maxCephPoolIdx

unsigned int g_maxCephPoolIdx = 1

size of the Striper/IoCtx pool, defaults to 1 may be overwritten in the configuration file (See XrdCephOss::configure)

Definition at line 95 of file XrdCephPosix.cc.

Referenced by ceph_posix_disconnect_all(), XrdCephOss::Configure(), and getCephPoolIdxAndIncrease().

g_namelib

XrdOucName2Name* g_namelib = 0

pointer to library providing Name2Name interface. 0 be default populated in case of ceph.namelib entry in the config file in XrdCephOss

Definition at line 98 of file XrdCephPosix.cc.

Referenced by XrdCephOss::Configure(), m_translateFileName(), and translateFileName().

g_nextCephFd

unsigned int g_nextCephFd = 0

global variable remembering the next available file descriptor

Definition at line 105 of file XrdCephPosix.cc.

Referenced by insertFileRef().

g_radosStripers

std::vector<StriperDict> g_radosStripers

Definition at line 81 of file XrdCephPosix.cc.

Referenced by ceph_posix_disconnect_all(), checkAndCreateStriper(), dumpClusterInfo(), getCephPoolIdxAndIncrease(), and getRadosStriper().

g_striper_mutex

XrdSysMutex g_striper_mutex

mutex protecting the striper and ioctx maps

Definition at line 86 of file XrdCephPosix.cc.

Referenced by ceph_posix_disconnect_all(), getIoCtx(), and getRadosStriper().