XrdCeph Review 26/04/23

XrdOssDF* XrdCephOss::newFile(const char *tident) {

  // Depending on the configuration settings stack up the underlying 
  // XrdCephOssFile instance with decorator objects for readV and Buffering requests

  XrdCephOssFile* xrdCephOssDF =  new XrdCephOssFile(this);
  
  if (m_configReadVEnable) {
    xrdCephOssDF = new XrdCephOssReadVFile(this,xrdCephOssDF,m_configReadVAlgName);
  }

  if (m_configBufferEnable) {
    xrdCephOssDF = new XrdCephOssBufferedFile(this,xrdCephOssDF, m_configBufferSize, 
                                              m_configBufferIOmode, m_configMaxSimulBufferCount);
  }


  return xrdCephOssDF;
}

       if (!strncmp(var, "ceph.bufferiomode", 17)) {
         var = Config.GetWord();
         if (var) {
           // Warn in case parameters were givne
           char parms[1040];
           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
         } else {
           Eroute.Emsg("Config", "Missing value for ceph.bufferiomode in config file", configfn);
           return 1;
         }
       }

/// 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
unsigned int getCephPoolIdxAndIncrease() {
  if (g_radosStripers.size() == 0) {
    // make sure we do not have a race condition here
    XrdSysMutexHelper lock(g_init_mutex);
    // double check now that we have the lock
    if (g_radosStripers.size() == 0) {
      // initialization phase : allocate corresponding places in the vectors
      for (unsigned int i = 0; i < g_maxCephPoolIdx; i++) {
        g_radosStripers.push_back(StriperDict());
        g_ioCtx.push_back(IOCtxDict());
        g_cluster.push_back(0);
      }
    }
  }