diff --git a/.github/workflows/CI.yml b/.github/workflows/CI.yml index e3dc0a26721..55b508a7036 100644 --- a/.github/workflows/CI.yml +++ b/.github/workflows/CI.yml @@ -92,7 +92,6 @@ jobs: run: | tests/post-install.sh tests/check-headers.sh - alma8: name: Alma 8 runs-on: ubuntu-latest diff --git a/VERSION b/VERSION index f4034a5d279..b9260c78903 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -$Format:%(describe)$ +v5.9.4 diff --git a/packaging/makesrpm.sh b/packaging/makesrpm.sh index 208f7f5c6ec..a2e2e59a366 100755 --- a/packaging/makesrpm.sh +++ b/packaging/makesrpm.sh @@ -125,6 +125,11 @@ echo "[i] Working with version: $VERSION" # Sanitize version to work with RPMs # https://docs.fedoraproject.org/en-US/packaging-guidelines/Versioning/ #------------------------------------------------------------------------------- +RELEASE=4 +if test x`echo $VERSION | grep -E $RCEXP` != x; then + RELEASE=0.`echo $VERSION | sed 's/.*-rc/rc/'` + VERSION=`echo $VERSION | sed 's/-rc.*//'` +fi VERSION=${VERSION#v} # remove "v" prefix VERSION=${VERSION/-rc/~rc} # release candidates use ~ in RPMs diff --git a/src/XrdCeph/CMakeLists.txt b/src/XrdCeph/CMakeLists.txt index 2619b934c35..cfe5e7474a0 100644 --- a/src/XrdCeph/CMakeLists.txt +++ b/src/XrdCeph/CMakeLists.txt @@ -13,12 +13,15 @@ else() endif() endif() +include_directories( ${RADOS_INCLUDE_DIR} ) + set(BUILD_CEPH TRUE CACHE BOOL INTERNAL FORCE) add_library(XrdCephPosix SHARED XrdCephPosix.cc XrdCephPosix.hh XrdCephBulkAioRead.cc XrdCephBulkAioRead.hh) + target_compile_options(XrdCephPosix PRIVATE -Wno-deprecated-declarations) @@ -29,7 +32,7 @@ target_include_directories(XrdCephPosix PUBLIC ${RADOS_INCLUDE_DIR} $) set_target_properties(XrdCephPosix - PROPERTIES VERSION 0.0.1 SOVERSION 0) + PROPERTIES VERSION 5.7.0 SOVERSION 0) set(LIB_XRD_CEPH XrdCeph-${PLUGIN_VERSION}) @@ -50,7 +53,7 @@ add_library(${LIB_XRD_CEPH} MODULE ) target_link_libraries(${LIB_XRD_CEPH} - PRIVATE XrdCephPosix XrdUtils XrdServer ${RADOS_LIBS}) + PRIVATE ${XROOTD_LIBRARIES} XrdCephPosix XrdUtils XrdServer ${RADOS_LIBS}) set(LIB_XRD_CEPH_XATTR XrdCephXattr-${PLUGIN_VERSION}) diff --git a/src/XrdCeph/XrdCephOss.cc b/src/XrdCeph/XrdCephOss.cc index 5ff411d33f4..75f58d8d40b 100644 --- a/src/XrdCeph/XrdCephOss.cc +++ b/src/XrdCeph/XrdCephOss.cc @@ -28,14 +28,8 @@ #include #include -#include "XrdVersion.hh" -#include "XrdCeph/XrdCephOss.hh" -#include "XrdCeph/XrdCephOssDir.hh" -#include "XrdCeph/XrdCephOssFile.hh" +#include #include "XrdCeph/XrdCephPosix.hh" -#include "XrdCeph/XrdCephOssBufferedFile.hh" -#include "XrdCeph/XrdCephOssReadVFile.hh" - #include "XrdOuc/XrdOucEnv.hh" #include "XrdSys/XrdSysError.hh" #include "XrdSys/XrdSysPlatform.hh" @@ -43,6 +37,12 @@ #include "XrdOuc/XrdOucStream.hh" #include "XrdOuc/XrdOucName2Name.hh" #include "XrdOuc/XrdOucN2NLoader.hh" +#include "XrdVersion.hh" +#include "XrdCeph/XrdCephOss.hh" +#include "XrdCeph/XrdCephOssDir.hh" +#include "XrdCeph/XrdCephOssFile.hh" +#include "XrdCeph/XrdCephOssBufferedFile.hh" +#include "XrdCeph/XrdCephOssReadVFile.hh" XrdVERSIONINFO(XrdOssGetStorageSystem, XrdCephOss); @@ -115,7 +115,7 @@ ssize_t getNumericAttr(const char* const path, const char* attrName, const int m if (attrLen <= 0) { retval = -EINVAL; } else { - attrValue[attrLen] = '\0'; + attrValue[attrLen] = (char)'\0'; char *endPointer = (char *)NULL; retval = strtoll(attrValue, &endPointer, 10); } @@ -127,6 +127,9 @@ ssize_t getNumericAttr(const char* const path, const char* attrName, const int m return retval; } +char *g_cksLogFileName; + +extern FILE *g_cksLogFile; extern "C" { @@ -165,6 +168,12 @@ XrdCephOss::~XrdCephOss() { extern unsigned int g_maxCephPoolIdx; extern unsigned int g_cephAioWaitThresh; +extern bool g_calcStreamedAdler32; +extern bool g_storeStreamedAdler32; +extern bool g_logStreamedAdler32; +extern double g_ECcorrectionFactor; // correction factor to apply to used space when EC pools are used, to get a better estimate of actual used space + + int XrdCephOss::Configure(const char *configfn, XrdSysError &Eroute) { int NoGo = 0; XrdOucEnv myEnv; @@ -348,7 +357,7 @@ int XrdCephOss::Configure(const char *configfn, XrdSysError &Eroute) { if (!Config.GetRest(parms, sizeof(parms)) || parms[0]) { Eroute.Emsg("Config", "readvalgname parameters will be ignored"); } - m_configBufferIOmode = var; // allowed values would be aio, io + m_configBufferIOmode = var; // allowed values would be aio, io, write-only-io } else { Eroute.Emsg("Config", "Missing value for ceph.bufferiomode in config file", configfn); return 1; @@ -361,19 +370,97 @@ int XrdCephOss::Configure(const char *configfn, XrdSysError &Eroute) { m_configPoolnames = var; } else { Eroute.Emsg("Config", "Missing value for ceph.reportingpools in config file", configfn); - return 1; + return 1; + } + // EC correction factor for pool reporting + if (!strncmp(var, "ceph.ECcorrectionFactor", 23)) { // size in bytes + var = Config.GetWord(); + if (var) { + double value = strtod(var, 0); + if (value > 0 and value <= 1) { + g_ECcorrectionFactor = value; + Eroute.Emsg("Config", "ceph.ECcorrectionFactor", std::to_string(g_ECcorrectionFactor).c_str() ); + } else { + Eroute.Emsg("Config", "Invalid value for ceph.ECcorrectionFactor in config file; enter a value between 0 and 1", configfn, var); + return 1; + } + } else { + Eroute.Emsg("Config", "Missing value for ceph.ECcorrectionFactor in config file. Setting default 8/11", configfn); + g_ECcorrectionFactor = 0.727272; // default 8/11 EC correction factor + return 1; + } + } + } + + if (!strcmp(var, "ceph.streamed-cks-adler32")) { // Streaming Adler32 checksum + + var = Config.GetWord(); + if (var) { +/* + * Currently, actions are simply additive: + * + * Store implies calculate, log, store + * Log implies calculate, log + * Calc implies calculate + * + * Might want to make e.g. logging optional in the future, + * when storing is more prevalent. + * + * Instead of setting g_* flags in three conditionals, + * can switch to setting values in a single bitfield flag + * + */ + if (strstr(var, "calc")) { + g_calcStreamedAdler32 = true; + g_logStreamedAdler32 = false; + g_storeStreamedAdler32 = false; + } + if (strstr(var, "log")) { + g_calcStreamedAdler32 = true; + g_logStreamedAdler32 = true; + g_storeStreamedAdler32 = false; + } + if (strstr(var, "store")) { + g_calcStreamedAdler32 = true; + g_logStreamedAdler32 = true; + g_storeStreamedAdler32 = true; + } + } - } - } // while + }// "ceph.streamed-cks-adler32" - // Now check if any errors occurred during file i/o + if (!strcmp(var, "ceph.streamed-cks-logfile") ) { + var = Config.GetWord(); + if (var) { + g_cksLogFileName = strdup(var); + } else { + const char *defLogFileName = "/tmp/checksums.log"; // To-DO: Move defLogFileName so it can also be used as fallback + // when attempt to open specified log file below fails + Eroute.Emsg("Config", "Missing value for ceph.streamed-cks-logfile in config file, setting to default = ", defLogFileName); + g_cksLogFileName = strdup(defLogFileName); + return 1; + } + }// "ceph.streamed-cks-logfile" + } + // Now check if any errors occured during file i/o int retc = Config.LastError(); if (retc) { NoGo = Eroute.Emsg("Config", -retc, "read config file", configfn); } Config.Close(); + + if (g_logStreamedAdler32) { + if (NULL == (g_cksLogFile = fopen(g_cksLogFileName, "a"))) { + g_logStreamedAdler32 = false; + Eroute.Emsg("Config: ", "cannot open file for logging checksum values and pathname", g_cksLogFileName); + return 1; + } else { + Eroute.Emsg("Config: ", "Opened file for logging checksum values and pathname: ", g_cksLogFileName); + } + } + } return NoGo; } @@ -606,20 +693,16 @@ int XrdCephOss::StatLS(XrdOucEnv &env, const char *charPath, char *buff, int &bl XrdCephEroute.Say("Failed to get used space in pool ", spath.c_str()); return -EINVAL; } - // Construct the object path std::string spaceInfoPath = spath + ":" + (const char *)"__spaceinfo__"; totalSpace = getNumericAttr(spaceInfoPath.c_str(), "total_space", 24); if (totalSpace < 0) { XrdCephEroute.Say("Could not get 'total_space' attribute from ", spaceInfoPath.c_str()); return -EINVAL; - } - + } // // Figure for 'usedSpace' already accounts for Erasure Coding overhead // - - freeSpace = totalSpace - usedSpace; blen = formatStatLSResponse(buff, blen, spath.c_str(), /* "oss.cgroup" */ @@ -679,3 +762,4 @@ XrdOssDF* XrdCephOss::newFile(const char *tident) { return xrdCephOssDF; } + diff --git a/src/XrdCeph/XrdCephOss.hh b/src/XrdCeph/XrdCephOss.hh index d79743f8a71..a54040cb2cc 100644 --- a/src/XrdCeph/XrdCephOss.hh +++ b/src/XrdCeph/XrdCephOss.hh @@ -26,9 +26,7 @@ #define __CEPH_OSS_HH__ #include - #include -#include //------------------------------------------------------------------------------ //! This class implements XrdOss interface for usage with a CEPH storage. diff --git a/src/XrdCeph/XrdCephOssBufferedFile.cc b/src/XrdCeph/XrdCephOssBufferedFile.cc index 0862d39b4d0..7783ccc481d 100644 --- a/src/XrdCeph/XrdCephOssBufferedFile.cc +++ b/src/XrdCeph/XrdCephOssBufferedFile.cc @@ -151,6 +151,10 @@ ssize_t XrdCephOssBufferedFile::Read(off_t offset, size_t blen) { ssize_t XrdCephOssBufferedFile::Read(void *buff, off_t offset, size_t blen) { size_t thread_id = std::hash{}(std::this_thread::get_id()); + if (m_bufferIOmode == "write-only-io") { + return m_xrdOssDF->Read(buff, offset, blen); + } + IXrdCephBufferAlg * buffer{nullptr}; // check for, and create if needed, a buffer { @@ -326,11 +330,12 @@ std::unique_ptr XrdCephOssBufferedFile::create std::unique_ptr cephio; if (m_bufferIOmode == "aio") { cephio = std::unique_ptr(new CephIOAdapterAIORaw(cephbuffer.get(),m_fd)); - } else if (m_bufferIOmode == "io") { + } else if (m_bufferIOmode == "io" || m_bufferIOmode == "write-only-io") { cephio = std::unique_ptr(new CephIOAdapterRaw(cephbuffer.get(),m_fd, !m_cephoss->m_useDefaultPreadAlg)); } else { - BUFLOG("XrdCephOssBufferedFile: buffer mode needs to be one of aio|io " ); + BUFLOG("XrdCephOssBufferedFile: buffer mode needs to be one of aio|io|write-only-io " ); + m_xrdOssDF->Close(); return bufferAlg; // invalid instance; } diff --git a/src/XrdCeph/XrdCephPosix.cc b/src/XrdCeph/XrdCephPosix.cc index 65ee444a114..e052418b603 100644 --- a/src/XrdCeph/XrdCephPosix.cc +++ b/src/XrdCeph/XrdCephPosix.cc @@ -28,7 +28,7 @@ #include #include -#include +#include #include #include #include @@ -54,7 +54,49 @@ #include "XrdCeph/XrdCephPosix.hh" #include "XrdCeph/XrdCephBulkAioRead.hh" #include "XrdSfs/XrdSfsFlags.hh" // for the OFFLINE flag status +#include "XrdCks/XrdCksData.hh" +#include "XrdCks/XrdCksCalcadler32.hh" +#include +char *ts_rfc3339() { + + std::time_t now = std::time({}); + char timeString[std::size("yyyy-mm-dd hh:mm:ss")]; + std::strftime(std::data(timeString), std::size(timeString), + "%F %TZ", std::gmtime(&now)); + return strdup(timeString); +} + +constexpr char hex2ascii(char nibble) { return (0<= nibble && nibble<=9) ? nibble+'0' : nibble-10+'a'; } +constexpr char hiNibble(uint8_t hexbyte) { return (hexbyte & 0xf0) >> 4; } +constexpr char loNibble(uint8_t hexbyte) { return (hexbyte & 0x0f); } + +constexpr char *hexbytes2ascii(const char bytes[], const unsigned int length){ + + char asciiVal[9] {}; + for (unsigned int i = 0, j = 0; i < length; i++) { + + const uint8_t hexbyte = bytes[i]; + asciiVal[j++] = hex2ascii(hiNibble(hexbyte)); + asciiVal[j++] = hex2ascii(loNibble(hexbyte)); + + } + return strdup(asciiVal); +} + +using namespace std; + +int setXrdCksAttr(const int fd, const char* cstype, const char* ckSumbuf) { + + int rc = -1; + + std::vector attrData = XrdCksAttrData(cstype, ckSumbuf, time(0)); + + rc = ceph_posix_fsetxattr(fd, XrdCksAttrName(cstype).c_str(), + attrData.data(), attrData.size(), 0); + + return rc; +} /// small struct for directory listing struct DirIterator { @@ -107,10 +149,20 @@ unsigned int g_nextCephFd = 0; XrdSysMutex g_fd_mutex; /// mutex protecting initialization of ceph clusters XrdSysMutex g_init_mutex; +/// mutex protecting checksum log file +XrdSysMutex g_logAdler32; //JW Counter for number of times a given cluster is resolved. std::map g_idxCntr; +//IJJ: Actions for Adler32 checksum +bool g_calcStreamedAdler32; +bool g_logStreamedAdler32; +bool g_storeStreamedAdler32; +double g_ECcorrectionFactor; + +FILE *g_cksLogFile; + /// 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 @@ -529,6 +581,7 @@ int checkAndCreateStriper(unsigned int cephPoolIdx, std::string &userAtPool, con } int rc = g_cluster[cephPoolIdx]->ioctx_create(file.pool.c_str(), *ioctx); if (rc != 0) { + logwrapper((char*)"checkAndCreateStriper : ioctx_create failed, user@pool = %s", userAtPool.c_str()); logwrapper((char*)"checkAndCreateStriper : ioctx_create failed, rc = %d", rc); cluster->shutdown(); delete cluster; @@ -588,8 +641,9 @@ int checkAndCreateStriper(unsigned int cephPoolIdx, std::string &userAtPool, con return 0; } IOCtxDict & ioDict = g_ioCtx[cephPoolIdx]; - ioDict.emplace(userAtPool, ioctx); - sDict.emplace(userAtPool, striper); + ioDict.insert(std::pair(userAtPool, ioctx)); + sDict.insert(std::pair + (userAtPool, striper)).first; } return 1; } @@ -664,7 +718,7 @@ static int ceph_posix_internal_truncate(const CephFile &file, unsigned long long int ceph_posix_open(XrdOucEnv* env, const char *pathname, int flags, mode_t mode){ CephFileRef fr = getCephFileRef(pathname, env, flags, mode, 0); - + fr.writingData = false; struct stat buf; libradosstriper::RadosStriper *striper = getRadosStriper(fr); //Get a handle to the RADOS striper API if (NULL == striper) { @@ -748,7 +802,13 @@ int ceph_posix_open(XrdOucEnv* env, const char *pathname, int flags, mode_t mode } } } + // At this point, we know either the target file didn't exist, or the ceph_posix_unlink above removed it + if (g_calcStreamedAdler32) { + fr.cksCalcadler32 = new XrdCksCalcadler32(); + fr.cksCalcadler32->Init(); + } + fr.writingData = true; int fd = insertFileRef(fr); logwrapper((char*)"File descriptor %d associated to file %s opened in write mode", fd, pathname); return fd; @@ -757,6 +817,17 @@ int ceph_posix_open(XrdOucEnv* env, const char *pathname, int flags, mode_t mode } + +const char* formatAdler32(unsigned long adler32) { + +#ifndef Xrd_Big_Endian + adler32 = htonl(adler32); +#endif + char adler32Cks[8+1]; + sprintf(adler32Cks, "%08lx", adler32); + return (const char*)strdup(adler32Cks); +} + int ceph_posix_close(int fd) { CephFileRef* fr = getFileRef(fd); if (fr) { @@ -769,6 +840,9 @@ int ceph_posix_close(int fd) { lastAsyncAge = 1.0 * (now.tv_sec - fr->lastAsyncSubmission.tv_sec) + 0.000001 * (now.tv_usec - fr->lastAsyncSubmission.tv_usec); } + if (fr->bytesWritten > 0){ + ceph_posix_fremovexattr(fd,"XrdCks.adler32"); + } 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, " @@ -777,6 +851,32 @@ int ceph_posix_close(int fd) { fr->asyncWrCompletionCount, fr->asyncWrStartCount, fr->bytesAsyncWritePending, fr->asyncRdCompletionCount, fr->asyncRdStartCount, fr->bytesWritten, fr->maxOffsetWritten, fr->longestAsyncWriteTime, fr->longestCallbackInvocation, (lastAsyncAge)); + + if (fr->writingData && g_calcStreamedAdler32) { + + unsigned long adlerULong; + memcpy((&adlerULong), fr->cksCalcadler32->Final(), 4); + const char* adler32Cks = formatAdler32(adlerULong); + + logwrapper((char*)"ceph_close: fd: %d, Adler32 streamed checksum = %s", fd, adler32Cks); + + if (g_logStreamedAdler32) { + const char *path = strdup((fr->pool + ":" + fr->name).c_str()); + XrdSysMutexHelper lock(g_logAdler32); + fprintf(g_cksLogFile, "%s,%s,%s,%s,%s\n", ts_rfc3339(), path, "streamed", "adler32", adler32Cks); + fflush(g_cksLogFile); + } + + + if (g_storeStreamedAdler32) { + int rc = setXrdCksAttr(fd, "adler32", adler32Cks); + if (rc != 0) { + logwrapper((char*)"ceph_close: Can't set attribute XrdCks.adler32 for checksum"); + } + } + delete fr->cksCalcadler32; + } + deleteFileRef(fd, *fr); return 0; } else { @@ -838,6 +938,9 @@ ssize_t ceph_posix_write(int fd, const void *buf, size_t count) { fr->wrcount++; fr->bytesWritten+=count; if (fr->offset) fr->maxOffsetWritten = std::max(fr->offset - 1, fr->maxOffsetWritten); + if (g_calcStreamedAdler32) { + fr->cksCalcadler32->Update((const char*)buf, count); + } return count; } else { return -EBADF; @@ -848,6 +951,7 @@ 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; @@ -859,11 +963,16 @@ ssize_t ceph_posix_pwrite(int fd, const void *buf, size_t count, off64_t offset) 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); + fr->bytesWritten+=count; + if (offset + count) fr->maxOffsetWritten = std::max(offset + count - 1, fr->maxOffsetWritten); + if (g_calcStreamedAdler32) { + fr->cksCalcadler32->Update((const char*)buf, count); + } + if (offset + count) fr->maxOffsetWritten = std::max(offset + count - 1, fr->maxOffsetWritten); return count; } else { return -EBADF; @@ -882,7 +991,7 @@ static void ceph_aio_write_complete(rados_completion_t c, void *arg) { 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)); + fr->maxOffsetWritten = std::max(fr->maxOffsetWritten, 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); @@ -908,6 +1017,7 @@ ssize_t ceph_aio_write(int fd, XrdSfsAio *aiop, AioCB *cb) { 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; @@ -938,6 +1048,10 @@ ssize_t ceph_aio_write(int fd, XrdSfsAio *aiop, AioCB *cb) { fr->asyncWrStartCount++; ::gettimeofday(&fr->lastAsyncSubmission, nullptr); fr->bytesAsyncWritePending+=count; + // disable streamed checksums for aio writes as it's too risky + //if (g_calcStreamedAdler32) { + // fr->cksCalcadler32->Update((const char*)buf, count); + // } return rc; } else { return -EBADF; @@ -1349,8 +1463,23 @@ static ssize_t ceph_posix_internal_setxattr(const CephFile &file, const char* na ssize_t ceph_posix_setxattr(XrdOucEnv* env, const char* path, const char* name, const void* value, size_t size, int flags) { + int rc; + + auto *cks = (XrdCksData*)value; 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); + rc = ceph_posix_internal_setxattr(getCephFile(path, env), name, value, size, flags); + + if (0 == rc && !strcmp(name, "XrdCks.adler32") && g_logStreamedAdler32) { +// +// We know that streamed checksums use ceph_posix_fsetxattr below, so this must be a readback checksum +// + auto cksAscii = (const char*)hexbytes2ascii(cks->Value, cks->Length); + logwrapper((char*)"readback checksum = %s", cksAscii); + fprintf(g_cksLogFile, "%s,%s,%s,%s,%s\n", ts_rfc3339(), path, "readback", "adler32", cksAscii); + fflush(g_cksLogFile); + + } + return rc; } int ceph_posix_fsetxattr(int fd, @@ -1508,7 +1637,7 @@ int ceph_posix_stat_pool(char const *poolName, long long *usedSpace) { } else { - *usedSpace = stat[poolName].num_kb * 1024; + *usedSpace = stat[poolName].num_kb * 1024 * g_ECcorrectionFactor; // num_kb is in KB, convert to bytes and apply EC correction factor return XrdOssOK; } @@ -1543,7 +1672,6 @@ 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); @@ -1627,3 +1755,4 @@ int ceph_posix_closedir(DIR *dirp) { delete ((DirIterator*)dirp); return 0; } + diff --git a/src/XrdCeph/XrdCephPosix.hh b/src/XrdCeph/XrdCephPosix.hh index 5d01129b6ae..1dfee919527 100644 --- a/src/XrdCeph/XrdCephPosix.hh +++ b/src/XrdCeph/XrdCephPosix.hh @@ -39,6 +39,7 @@ #include "XrdSys/XrdSysPthread.hh" #include "XrdOuc/XrdOucIOVec.hh" +#include // simple logging for XrdCeph buffering code #define XRDCEPHLOGLEVEL 1 @@ -107,6 +108,7 @@ struct CephFile { unsigned int nbStripes; unsigned long long stripeUnit; unsigned long long objectSize; + }; struct CephFileRef : CephFile { @@ -127,6 +129,8 @@ struct CephFileRef : CephFile { ::timeval lastAsyncSubmission; double longestAsyncWriteTime; double longestCallbackInvocation; + bool writingData; + XrdCksCalcadler32 *cksCalcadler32; }; #endif // __XRD_CEPH_POSIX__ diff --git a/src/XrdCeph/XrdCephXAttr.cc b/src/XrdCeph/XrdCephXAttr.cc index 4c2f60a0f96..b56a6fc5b03 100644 --- a/src/XrdCeph/XrdCephXAttr.cc +++ b/src/XrdCeph/XrdCephXAttr.cc @@ -28,6 +28,9 @@ #include "XrdOuc/XrdOucTrace.hh" #include "XrdCeph/XrdCephXAttr.hh" +#include "XrdCks/XrdCksData.hh" +#include + XrdSysError XrdCephXattrEroute(0); XrdOucTrace XrdCephXattrTrace(&XrdCephXattrEroute); @@ -49,6 +52,7 @@ extern "C" XrdCephXattrEroute.Say("CephXattr loading failed with exception. Check the syntax of parameters : ", parms); return 0; } + return new XrdCephXAttr(); } } @@ -99,16 +103,19 @@ int XrdCephXAttr::List(AList **aPL, const char *Path, int fd, int getSz) { int XrdCephXAttr::Set(const char *Aname, const void *Aval, int Avsz, const char *Path, int fd, int isNew) { + int rc = 0; + if (fd >= 0) { - return ceph_posix_fsetxattr(fd, Aname, Aval, Avsz, 0); + rc = ceph_posix_fsetxattr(fd, Aname, Aval, Avsz, 0); } else { try { - return ceph_posix_setxattr(0, Path, Aname, Aval, Avsz, 0); + rc = ceph_posix_setxattr(0, Path, Aname, Aval, Avsz, 0); } catch (std::exception &e) { XrdCephXattrEroute.Say("Set : invalid syntax in file parameters", Path); - return -EINVAL; + rc = -EINVAL; } } + return rc; } XrdVERSIONINFO(XrdSysGetXAttrObject, XrdCephXAttr); diff --git a/src/XrdCeph/XrdCephXAttr.hh b/src/XrdCeph/XrdCephXAttr.hh index ed6f4d39f21..0dfb9f5ec74 100644 --- a/src/XrdCeph/XrdCephXAttr.hh +++ b/src/XrdCeph/XrdCephXAttr.hh @@ -136,7 +136,7 @@ public: //! attribute value which may contain binary data. //! @param Path -> Path of the file whose attribute is to be set. //! @param fd -> If >=0 is the file descriptor of the opened subject file. - //! @param isNew When !0 then the attribute must not exist (i.e. new). + //! @param isnew When !0 then the attribute must not exist (i.e. new). //! Otherwise, if it does exist, the value is replaced. In //! either case, if it does not exist it should be created. //! diff --git a/src/XrdCks/XrdCksManager.cc b/src/XrdCks/XrdCksManager.cc index 844fe5a6373..e5a8def588c 100644 --- a/src/XrdCks/XrdCksManager.cc +++ b/src/XrdCks/XrdCksManager.cc @@ -463,8 +463,11 @@ int XrdCksManager::Del(const char *Pfn, XrdCksData &Cks) int XrdCksManager::Get(const char *Pfn, XrdCksData &Cks) { XrdOucXAttr xCS; - time_t MTime = 0; - int rc = 0, nFault = 0; + + // not checking stale checksums as ceph file modification times constantly refresh on read + //time_t MTime; + int rc=0, nFault=0; + // Determine which checksum to get (we will accept unsupported ones as well) // @@ -481,16 +484,14 @@ int XrdCksManager::Get(const char *Pfn, XrdCksData &Cks) nFault = strcmp(xCS.Attr.Cks.Name, Cks.Name); Cks = xCS.Attr.Cks; -// Verify the file. We do this with modification time unless that check -// has been turned off. -// - if (CksOpts & Cks_nomtchk) MTime = Cks.fmTime; - else if ((rc = ModTime(Pfn, MTime))) return rc; +// Verify the file +// not done as ceph mod times are weird +// if ((rc = ModTime(Pfn, MTime))) return rc; + // Return result // - return (Cks.fmTime != MTime || nFault - || Cks.Length > XrdCksData::ValuSize || Cks.Length <= 0 + return ( nFault || Cks.Length > XrdCksData::ValuSize || Cks.Length <= 0 ? -ESTALE : int(Cks.Length)); } diff --git a/src/XrdCms/XrdCmsCluster.cc b/src/XrdCms/XrdCmsCluster.cc index 7db85777fc5..4bcad78f293 100644 --- a/src/XrdCms/XrdCmsCluster.cc +++ b/src/XrdCms/XrdCmsCluster.cc @@ -36,6 +36,7 @@ #include #include #include +#include #include "XProtocol/YProtocol.hh" @@ -969,7 +970,7 @@ int XrdCmsCluster::Select(XrdCmsSelect &Sel) pmask |= pinfo.ssvec & Sel.nmask; if (pmask && maxBits(pmask, baseFS.dfsTries())) {Sel.Resp.DLen = snprintf(Sel.Resp.Data, sizeof(Sel.Resp.Data)-1, - "Too many DFS %s attempts; operation terminated", Amode)+1; + "File not found")+1; return RetryErr; } } @@ -1122,6 +1123,7 @@ int XrdCmsCluster::Select(SMask_t pmask, int &port, char *hbuff, int &hlen, : SelbyLoadR(pmask, selR)); if (nP) hlen = nP->netIF.GetName(hbuff, port, nType) + 1; + else hlen = 0; STMutex.UnLock(); return hlen != 1; @@ -1793,10 +1795,17 @@ XrdCmsNode *XrdCmsCluster::SelbyLoad(SMask_t mask, XrdCmsSelector &selR) XrdCmsNode *np, *sp = 0; bool Multi = false, reqSS = (selR.needSpace & XrdCmsNode::allowsSS) != 0; -// Scan for a node (preset possible, suspended, overloaded, full, and dead) -// - selR.Reset(); SelTcnt++; - for (int i = 0; i <= STHi; i++) + // Scan for a node (preset possible, suspended, overloaded, full, and dead) + // + selR.Reset(); SelTcnt++; + int selCap = 1; + int randomSel=1; + bool useWR=(Config.P_randlb==1); + //default 0 to skip the node in random selection if the below checks fail + int Weighed[STMax] = { 0 }; + //float scalingFactor = 1+(1-float(std::clamp(Config.P_fuzz,0,100))/100)*2; + for (int i = 0; i <= STHi; i++) + // Weighed[i] = 0; if ((np = NodeTab[i]) && (np->NodeMask & mask)) {if (!(selR.needNet & np->hasNet)) {selR.xNoNet= true; continue;} selR.nPick++; @@ -1807,7 +1816,14 @@ XrdCmsNode *XrdCmsCluster::SelbyLoad(SMask_t mask, XrdCmsSelector &selR) || (reqSS && np->isNoStage))) {selR.xFull = true; continue;} if (!sp) sp = np; - else{if (selR.needSpace) + else{ + if (useWR){ + //add 1 to the inverse load, this is to allow some selection in case reported loads hit 100 + int nload = 101 - np->myLoad; + selCap += nload;//static_cast(nload + std::pow(nload, scalingFactor)/2); + Weighed[i] = selCap; + } + else{if (selR.needSpace) {if (abs(sp->myMass - np->myMass) <= Config.P_fuzz) {if (sp->RefW > (np->RefW+Config.DiskLinger)) sp=np;} else if (sp->myMass > np->myMass) sp=np; @@ -1823,8 +1839,22 @@ XrdCmsNode *XrdCmsCluster::SelbyLoad(SMask_t mask, XrdCmsSelector &selR) } Multi = true; } + } } - + if (useWR){ + // pick a random weighed node + // + static std::random_device rand_dev; + static std::default_random_engine generator(rand_dev()); + static std::uniform_int_distribution distr(randomSel,selCap); + randomSel = distr(generator); + for(int i=0;i<=STHi;i++){ + if(randomSel<=Weighed[i]){ + sp=NodeTab[i]; + break; + } + } + } // Check for overloaded node and return result // if (!sp) return calcDelay(selR); @@ -1850,9 +1880,9 @@ XrdCmsNode *XrdCmsCluster::SelbyLoadR(SMask_t mask, XrdCmsSelector &selR) selR.Reset(); SelTcnt++; - int totWeight = 0; + for (int i = 0; i <= STHi; ++i) { NodeWeight[i] = 0; // make node unselectable first diff --git a/src/XrdCms/XrdCmsCluster.hh b/src/XrdCms/XrdCmsCluster.hh index 4b16d1f7949..3a30145c258 100644 --- a/src/XrdCms/XrdCmsCluster.hh +++ b/src/XrdCms/XrdCmsCluster.hh @@ -246,7 +246,6 @@ static const int AltSize = 254; // We may revert to IP address XrdSysRWLock STMutex; // Protects all node information variables XrdCmsNode *NodeTab[STMax]; // Current set of nodes int NodeWeight[STMax]; // Current set of load balancing weights - int STHi; // NodeTab high watermark int Reserved; RAtomic_llong SelWtot; // Total number of r/w selections (successful) diff --git a/src/XrdCms/XrdCmsConfig.cc b/src/XrdCms/XrdCmsConfig.cc index 3a2a69af3bb..9a46f235955 100644 --- a/src/XrdCms/XrdCmsConfig.cc +++ b/src/XrdCms/XrdCmsConfig.cc @@ -736,6 +736,7 @@ void XrdCmsConfig::ConfigDefaults(void) P_load = 0; P_mem = 0; P_pag = 0; + P_randlb = 0; // SelbyLoad algorithm choice AskPerf = 10; // Every 10 pings AskPing = 60; // Every 1 minute PingTick = 0; @@ -2688,7 +2689,8 @@ int XrdCmsConfig::xsched(XrdSysError *eDest, XrdOucStream &CFile) {"refreset", -1, &RefReset}, {"affinity", -2, 0}, {"affpath", -3, 0}, - {"tryhname", 1, &V_hntry} + {"tryhname", 1, &V_hntry}, + {"randlb", 1, &P_randlb} }; int numopts = sizeof(scopts)/sizeof(struct schedopts); diff --git a/src/XrdCms/XrdCmsConfig.hh b/src/XrdCms/XrdCmsConfig.hh index 5f925a4a0b0..daae032a0c9 100644 --- a/src/XrdCms/XrdCmsConfig.hh +++ b/src/XrdCms/XrdCmsConfig.hh @@ -107,6 +107,7 @@ int P_io; // % I/O Capacity in load factor int P_load; // % MSC Capacity in load factor int P_mem; // % MEM Capacity in load factor int P_pag; // % PAG Capacity in load factor +int P_randlb; // enable weighted random load balancing char DoMWChk; // When true (default) perform multiple write check char DoHnTry; // When true (default) use hostnames for try redirs diff --git a/src/XrdMacaroons/XrdMacaroonsConfigure.cc b/src/XrdMacaroons/XrdMacaroonsConfigure.cc index 8581d3c1791..2f47530904c 100644 --- a/src/XrdMacaroons/XrdMacaroonsConfigure.cc +++ b/src/XrdMacaroons/XrdMacaroonsConfigure.cc @@ -180,15 +180,16 @@ bool Handler::xsitename(XrdOucStream &config_obj, XrdSysError *log, std::string bool Handler::xsecretkey(XrdOucStream &config_obj, XrdSysError *log, std::string &secret) { char *val = config_obj.GetWord(); - - if (!val || !val[0]) { + if (!val || !val[0]) + { log->Emsg("Config", "Shared secret key not specified"); return false; } - BIO *bio = BIO_new_file(val, "rb"); - if (!bio) { - log->Emsg("Config", "Failed to open shared secret key file", val); + FILE *fp = fopen(val, "rb"); + + if (fp == nullptr) { + log->Emsg("Config", errno, "open shared secret key file", val); return false; } @@ -208,11 +209,8 @@ bool Handler::xsecretkey(XrdOucStream &config_obj, XrdSysError *log, std::string } BIO_push(b64, bio); - - int inlen; - char inbuf[512]; - - while ((inlen = BIO_read(b64, inbuf, 512)) > 0) { + while ((inlen = BIO_read(b64, inbuf, 512)) > 0) + { if (inlen < 0) { if (errno == EINTR) continue; break; @@ -220,16 +218,12 @@ bool Handler::xsecretkey(XrdOucStream &config_obj, XrdSysError *log, std::string BIO_write(bio_out, inbuf, inlen); } } - if (inlen < 0) { - BIO_free_all(bio_out); BIO_free_all(b64); log->Emsg("Config", errno, "read secret key."); return false; } - if (!BIO_flush(bio_out)) { - BIO_free_all(bio_out); BIO_free_all(b64); log->Emsg("Config", errno, "flush secret key."); return false; @@ -237,6 +231,7 @@ bool Handler::xsecretkey(XrdOucStream &config_obj, XrdSysError *log, std::string char *decoded; long data_len = BIO_get_mem_data(bio_out, &decoded); + BIO_free_all(b64); secret = std::string(decoded, data_len); diff --git a/src/XrdXrootd/XrdXrootdConfig.cc b/src/XrdXrootd/XrdXrootdConfig.cc index 6be6bcd209d..fc0d7333aad 100644 --- a/src/XrdXrootd/XrdXrootdConfig.cc +++ b/src/XrdXrootd/XrdXrootdConfig.cc @@ -1690,7 +1690,8 @@ int XrdXrootdProtocol::xred(XrdOucStream &Config) {"rm", RD_rm}, {"rmdir", RD_rmdir}, {"stat", RD_stat}, - {"trunc", RD_trunc} + {"trunc", RD_trunc}, + {"write", RD_write} }; static const int rHLen = 264; char rHost[2][rHLen], *hP[2], *val; diff --git a/src/XrdXrootd/XrdXrootdProtocol.cc b/src/XrdXrootd/XrdXrootdProtocol.cc index c58bc60a551..c0b448fe5d7 100644 --- a/src/XrdXrootd/XrdXrootdProtocol.cc +++ b/src/XrdXrootd/XrdXrootdProtocol.cc @@ -471,7 +471,7 @@ int XrdXrootdProtocol::Process2() switch(Request.header.requestid) {case kXR_login: return do_Login(); case kXR_protocol: return do_Protocol(); - case kXR_bind: return do_Bind(); + //case kXR_bind: return do_Bind(); default: Response.Send(kXR_InvalidRequest, "Invalid request; user not logged in"); return Link->setEtext("request without login"); diff --git a/src/XrdXrootd/XrdXrootdProtocol.hh b/src/XrdXrootd/XrdXrootdProtocol.hh index b60ea4e1405..4b7e5aae6e0 100644 --- a/src/XrdXrootd/XrdXrootdProtocol.hh +++ b/src/XrdXrootd/XrdXrootdProtocol.hh @@ -241,7 +241,8 @@ private: enum RD_func {RD_chmod = 0, RD_chksum, RD_dirlist, RD_locate, RD_mkdir, RD_mv, RD_prepare, RD_prepstg, RD_rm, RD_rmdir, RD_stat, RD_trunc, RD_ovld, RD_client, - RD_open1, RD_open2, RD_open3, RD_open4, RD_Num}; + RD_write, RD_open1, RD_open2, RD_open3, RD_open4, + RD_Num}; int do_Auth(); int do_Bind(); diff --git a/src/XrdXrootd/XrdXrootdXeq.cc b/src/XrdXrootd/XrdXrootdXeq.cc index adc2e37a0e1..808a292271a 100644 --- a/src/XrdXrootd/XrdXrootdXeq.cc +++ b/src/XrdXrootd/XrdXrootdXeq.cc @@ -1461,15 +1461,22 @@ int XrdXrootdProtocol::do_Open() // doDig = (digFS && SFS_LCLPATH(fn)); -// Validate the path and then check if static redirection applies +// Validate the path/req type and then check if static redirection applies // if (doDig) {popt = XROOTDXP_NOLK; opC = 0;} - else {int ropt; - if (!(popt = Squash(fn))) return vpEmsg("Opening", fn); - if (Route[RD_open1].Host[rdType] && (ropt = RPList.Validate(fn))) - return Response.Send(kXR_redirect, Route[ropt].Port[rdType], - Route[ropt].Host[rdType]); - } + else {int ropt = -1; + if (!(popt = Squash(fn))) return vpEmsg("Opening", fn); + if (Route[RD_open1].Host[rdType]) + ropt = RPList.Validate(fn); + else + if (Route[RD_write].Host[rdType] && ('w' == usage || strchr(op, 'd'))) + ropt = RD_write; + if (ropt > 0) + return Response.Send( + kXR_redirect, Route[ropt].Port[rdType], + Route[ropt].Host[rdType] + ); + } // Add the multi-write option if this path supports it // diff --git a/src/XrdXrootd/XrdXrootdXeqPgrw.cc b/src/XrdXrootd/XrdXrootdXeqPgrw.cc index 48844b4b136..71f017bc70d 100644 --- a/src/XrdXrootd/XrdXrootdXeqPgrw.cc +++ b/src/XrdXrootd/XrdXrootdXeqPgrw.cc @@ -281,7 +281,6 @@ int XrdXrootdProtocol::do_PgRIO() iov[i++].iov_len = pgPageSize; buff += pgPageSize; } - // If this is an unaligned read, offset the unaligned segment in the buffer // so that remaining pages are page-aligned. It will be reset when needed. // We also calculate the actual length of the first read. diff --git a/xrootd.spec b/xrootd.spec index 53c84f1efa9..6eba9d07b14 100644 --- a/xrootd.spec +++ b/xrootd.spec @@ -9,7 +9,7 @@ Name: xrootd Epoch: 1 -Release: 1%{?dist}%{?with_clang:.clang}%{?with_asan:.asan} +Release: 2%{?dist}%{?with_clang:.clang}%{?with_asan:.asan}%{?with_openssl11:.ssl11} Summary: Extended ROOT File Server Group: System Environment/Daemons License: LGPL-3.0-or-later AND BSD-2-Clause AND BSD-3-Clause AND curl AND MIT AND Zlib @@ -273,12 +273,12 @@ xrdcl-http is an XRootD client plugin which allows XRootD to interact with HTTP repositories. %if %{with ceph} -%package ceph +%package ceph-buffered Summary: XRootD plugin for interfacing with the Ceph storage platform Group: System Environment/Libraries Requires: %{name}-libs%{?_isa} = %{epoch}:%{version}-%{release} -%description ceph +%description ceph-buffered The xrootd-ceph is an OSS layer plugin for the XRootD server for interfacing with the Ceph storage platform. %endif @@ -690,7 +690,7 @@ fi %config(noreplace) %{_sysconfdir}/%{name}/client.plugins.d/xrdcl-http-plugin.conf %if %{with ceph} -%files ceph +%files ceph-buffered %{_libdir}/libXrdCeph-5.so %{_libdir}/libXrdCephXattr-5.so %{_libdir}/libXrdCephPosix.so.*