XrdPfcConfiguration.cc

Go to the documentation of this file.

#include "XrdPfc.hh"
#include "XrdPfcTrace.hh"
#include "XrdPfcInfo.hh"
 
#include "XrdPfcResourceMonitor.hh"
#include "XrdPfcPurgePin.hh"
 
#include "XrdOss/XrdOss.hh"
 
#include "XrdOuc/XrdOucEnv.hh"
#include "XrdOuc/XrdOucUtils.hh"
#include "XrdOuc/XrdOucStream.hh"
#include "XrdOuc/XrdOucPinLoader.hh"
#include "XrdOuc/XrdOuca2x.hh"
 
#include "XrdVersion.hh"
#include "XrdOfs/XrdOfsConfigPI.hh"
#include "XrdSys/XrdSysXAttr.hh"
 
#include <fcntl.h>
 
extern XrdSysXAttr *XrdSysXAttrActive;
 
namespace XrdPfc
{
   const char *trace_what_strings[] = {"","error ","warning ","info ","debug ","dump "};
}
 
using namespace XrdPfc;
 
XrdVERSIONINFO(XrdOucGetCache, XrdPfc);
 
Configuration::Configuration() :
   m_hdfsmode(false),
   m_allow_xrdpfc_command(false),
   m_data_space("public"),
   m_meta_space("public"),
   m_diskTotalSpace(-1),
   m_diskUsageLWM(-1),
   m_diskUsageHWM(-1),
   m_fileUsageBaseline(-1),
   m_fileUsageNominal(-1),
   m_fileUsageMax(-1),
   m_purgeInterval(300),
   m_purgeColdFilesAge(-1),
   m_purgeAgeBasedPeriod(10),
   m_accHistorySize(20),
   m_dirStatsInterval(900),
   m_dirStatsStoreDepth(1),
   m_bufferSize(128*1024),
   m_RamAbsAvailable(0),
   m_RamKeepStdBlocks(0),
   m_wqueue_blocks(16),
   m_wqueue_threads(4),
   m_prefetch_max_blocks(10),
   m_hdfsbsize(128*1024*1024),
   m_flushCnt(2000),
   m_cs_UVKeep(-1),
   m_cs_Chk(CSChk_Net),
   m_cs_ChkTLS(false),
   m_onlyIfCachedMinSize(1024*1024),
   m_onlyIfCachedMinFrac(1.0)
{}
 
 
bool Cache::cfg2bytes(const std::string &str, long long &store, long long totalSpace, const char *name) const
{
   char errStr[1024];
   snprintf(errStr, 1024, "ConfigParameters() Error parsing parameter %s", name);
 
   if (::isalpha(*(str.rbegin())))
   {
      if (XrdOuca2x::a2sz(m_log, errStr, str.c_str(), &store, 0, totalSpace))
      {
         return false;
      }
   }
   else
   {
      char *eP;
      errno = 0;
      double frac = strtod(str.c_str(), &eP);
      if (errno || eP == str.c_str())
      {
         m_log.Emsg(errStr, str.c_str());
         return false;
      }
 
      store = static_cast<long long>(totalSpace * frac + 0.5);
   }
 
   if (store < 0 || store > totalSpace)
   {
     snprintf(errStr, 1024, "ConfigParameters() Error: parameter %s should be between 0 and total available disk space (%lld) - it is %lld (given as %s)",
              name, totalSpace, store, str.c_str());
     m_log.Emsg(errStr, "");
     return false;
   }
 
   return true;
}
 
bool Cache::blocksize_str2value(const char *from, const char *str,
                                long long &val, long long min, long long max) const
{
   if (XrdOuca2x::a2sz(m_log, "Error parsing block-size", str, &val, min, max))
      return false;
 
   if (val & 0xFFF) {
      val &= ~0x0FFF;
      val +=  0x1000;
      m_log.Emsg(from, "blocksize must be a multiple of 4 kB. Rounded up.");
   }
 
   return true;
}
 
bool Cache::prefetch_str2value(const char *from, const char *str,
                               int &val, int min, int max) const
{
   if (XrdOuca2x::a2i(m_log, "Error parsing prefetch block count", str, &val, min, max))
      return false;
 
   return true;
}
 
/* Function: xcschk
 
   Purpose:  To parse the directive: cschk <parms>
 
             parms:  [[no]net] [[no]tls] [[no]cache] [uvkeep <arg>]
 
             all     Checksum check on cache & net transfers.
             cache   Checksum check on cache only, 'no' turns it off.
             net     Checksum check on net transfers 'no' turns it off.
             tls     use TLS if server doesn't support checksums 'no' turns it off.
             uvkeep  Maximum amount of time a cached file make be kept if it
                     contains unverified checksums as n[d|h|m|s], where 'n'
                     is a non-negative integer. A value of 0 prohibits disk
                     caching unless the checksum can be verified. You can
                     also specify "lru" which means the standard purge policy
                     is to be used.
 
   Output: true upon success or false upon failure.
 */
bool Cache::xcschk(XrdOucStream &Config)
{
   const char *val, *val2;
   struct cschkopts {const char *opname; int opval;} csopts[] =
   {
      {"off",     CSChk_None},
      {"cache",   CSChk_Cache},
      {"net",     CSChk_Net},
      {"tls",     CSChk_TLS}
   };
   int i, numopts = sizeof(csopts)/sizeof(struct cschkopts);
   bool isNo;
 
   if (! (val = Config.GetWord()))
   {m_log.Emsg("Config", "cschk parameter not specified"); return false; }
 
   while(val)
   {
      if ((isNo = strncmp(val, "no", 2) == 0))
         val2 = val + 2;
      else
         val2 = val;
      for (i = 0; i < numopts; i++)
      {
         if (!strcmp(val2, csopts[i].opname))
         {
            if (isNo)
               m_configuration.m_cs_Chk &= ~csopts[i].opval;
            else if (csopts[i].opval)
               m_configuration.m_cs_Chk |= csopts[i].opval;
            else
               m_configuration.m_cs_Chk  = csopts[i].opval;
            break;
         }
      }
      if (i >= numopts)
      {
         if (strcmp(val, "uvkeep"))
         {
            m_log.Emsg("Config", "invalid cschk option -", val);
            return false;
         }
         if (!(val = Config.GetWord()))
         {
            m_log.Emsg("Config", "cschk uvkeep value not specified");
            return false;
         }
         if (!strcmp(val, "lru"))
            m_configuration.m_cs_UVKeep = -1;
         else
         {
            int uvkeep;
            if (XrdOuca2x::a2tm(m_log, "uvkeep time", val, &uvkeep, 0))
               return false;
            m_configuration.m_cs_UVKeep = uvkeep;
         }
      }
      val = Config.GetWord();
   }
   // Decompose into separate TLS state, it is only passed on to psx
   m_configuration.m_cs_ChkTLS =  m_configuration.m_cs_Chk & CSChk_TLS;
   m_configuration.m_cs_Chk   &= ~CSChk_TLS;
 
   m_env->Put("psx.CSNet", m_configuration.is_cschk_net() ? (m_configuration.m_cs_ChkTLS ? "2" : "1") : "0");
 
   return true;
}
 
 
/* Function: xdlib
 
   Purpose:  To parse the directive: decisionlib <path> [<parms>]
 
             <path>  the path of the decision library to be used.
             <parms> optional parameters to be passed.
 
 
   Output: true upon success or false upon failure.
 */
bool Cache::xdlib(XrdOucStream &Config)
{
   const char*  val;
 
   std::string libp;
   if (! (val = Config.GetWord()) || ! val[0])
   {
      TRACE(Info," Cache::Config() decisionlib not specified; always caching files");
      return true;
   }
   else
   {
      libp = val;
   }
 
   char params[4096];
   if (val[0])
      Config.GetRest(params, 4096);
   else
      params[0] = 0;
 
   XrdOucPinLoader* myLib = new XrdOucPinLoader(&m_log, 0, "decisionlib",
                                                libp.c_str());
 
   Decision *(*ep)(XrdSysError&);
   ep = (Decision *(*)(XrdSysError&))myLib->Resolve("XrdPfcGetDecision");
   if (! ep) {myLib->Unload(true); return false; }
 
   Decision * d = ep(m_log);
   if (! d)
   {
      TRACE(Error, "Config() decisionlib was not able to create a decision object");
      return false;
   }
   if (params[0])
      d->ConfigDecision(params);
 
   m_decisionpoints.push_back(d);
   return true;
}
 
/* Function: xplib
 
   Purpose:  To parse the directive: purgelib <path> [<parms>]
 
             <path>  the path of the decision library to be used.
             <parms> optional parameters to be passed.
 
 
   Output: true upon success or false upon failure.
 */
bool Cache::xplib(XrdOucStream &Config)
{
   const char*  val;
 
   std::string libp;
   if (! (val = Config.GetWord()) || ! val[0])
   {
      TRACE(Info," Cache::Config() purgelib not specified; will use LRU for purging files");
      return true;
   }
   else
   {
      libp = val;
   }
 
   char params[4096];
   if (val[0])
      Config.GetRest(params, 4096);
   else
      params[0] = 0;
 
   XrdOucPinLoader* myLib = new XrdOucPinLoader(&m_log, 0, "purgelib",
                                                libp.c_str());
 
   PurgePin *(*ep)(XrdSysError&);
   ep = (PurgePin *(*)(XrdSysError&))myLib->Resolve("XrdPfcGetPurgePin");
   if (! ep) {myLib->Unload(true); return false; }
 
   PurgePin * dp = ep(m_log);
   if (! dp)
   {
      TRACE(Error, "Config() purgelib was not able to create a Purge Plugin object?");
      return false;
   }
   m_purge_pin = dp;
 
   if (params[0])
      m_purge_pin->ConfigPurgePin(params);
 
 
   return true;
}
 
/* Function: xtrace
 
   Purpose:  To parse the directive: trace <level>
   Output: true upon success or false upon failure.
 */
bool Cache::xtrace(XrdOucStream &Config)
{
   char  *val;
   static struct traceopts {const char *opname; int opval; } tropts[] =
   {
      {"none",    0},
      {"error",   1},
      {"warning", 2},
      {"info",    3},
      {"debug",   4},
      {"dump",    5},
      {"dumpxl",  6}
   };
   int numopts = sizeof(tropts)/sizeof(struct traceopts);
 
   if (! (val = Config.GetWord()))
   {m_log.Emsg("Config", "trace option not specified"); return 1; }
 
   for (int i = 0; i < numopts; i++)
   {
      if (! strcmp(val, tropts[i].opname))
      {
         m_trace->What = tropts[i].opval;
         return true;
      }
   }
   m_log.Emsg("Config", "invalid trace option -", val);
   return false;
}
 
// Determine if oss spaces are operational and if they support xattrs.
bool Cache::test_oss_basics_and_features()
{
   static const char *epfx = "test_oss_basics_and_features()";
 
   const auto &conf = m_configuration;
   const char *user = conf.m_username.c_str();
   XrdOucEnv   env;
 
   auto check_space = [&](const char *space, bool &has_xattr)
   {
      std::string fname("__prerun_test_pfc_");
      fname += space;
      fname += "_space__";
      env.Put("oss.cgroup", space);
 
      int res = m_oss->Create(user, fname.c_str(), 0600, env, XRDOSS_mkpath);
      if (res != XrdOssOK) {
         m_log.Emsg(epfx, "Can not create a file on space", space);
         return false;
      }
      XrdOssDF *oss_file = m_oss->newFile(user);
      res = oss_file->Open(fname.c_str(), O_RDWR, 0600, env);
      if (res != XrdOssOK) {
         m_log.Emsg(epfx, "Can not open a file on space", space);
         return false;
      }
      res = oss_file->Write(fname.data(), 0, fname.length());
      if (res != (int) fname.length()) {
         m_log.Emsg(epfx, "Can not write into a file on space", space);
         return false;
      }
 
      has_xattr = true;
      long long fsize = fname.length();
      res = XrdSysXAttrActive->Set("pfc.fsize", &fsize, sizeof(long long), 0, oss_file->getFD(), 0);
      if (res != 0) {
         m_log.Emsg(epfx, "Can not write xattr to a file on space", space);
         has_xattr = false;
      }
 
      oss_file->Close();
 
      if (has_xattr) {
         char pfn[4096];
         m_oss->Lfn2Pfn(fname.c_str(), pfn, 4096);
         fsize = -1ll;
         res = XrdSysXAttrActive->Get("pfc.fsize", &fsize, sizeof(long long), pfn);
         if (res != sizeof(long long) || fsize != (long long) fname.length())
         {
            m_log.Emsg(epfx, "Can not read xattr from a file on space", space);
            has_xattr = false;
         }
      }
 
      res = m_oss->Unlink(fname.c_str());
      if (res != XrdOssOK) {
         m_log.Emsg(epfx, "Can not unlink a file on space", space);
         return false;
      }
 
      return true;
   };
 
   bool aOK = true;
   aOK &= check_space(conf.m_data_space.c_str(), m_dataXattr);
   aOK &= check_space(conf.m_meta_space.c_str(), m_metaXattr);
 
   return aOK;
}
 
//______________________________________________________________________________
/* Function: Config
 
   Purpose: To parse configuration file and configure Cache instance.
   Output:  true upon success or false upon failure.
 */
bool Cache::Config(const char *config_filename, const char *parameters, XrdOucEnv *env)
{
   // Indicate whether or not we are a client instance
   const char *theINS = getenv("XRDINSTANCE");
   m_isClient = (theINS != 0 && strncmp("*client ", theINS, 8) == 0);
 
   // Tell everyone else we are a caching proxy
   XrdOucEnv::Export("XRDPFC", 1);
 
   XrdOucEnv emptyEnv;
   XrdOucEnv *myEnv = env ? env : &emptyEnv;
 
   XrdOucStream Config(&m_log, theINS, myEnv, "=====> ");
 
   if (! config_filename || ! *config_filename)
   {
      TRACE(Error, "Config() configuration file not specified.");
      return false;
   }
 
   int fd;
   if ( (fd = open(config_filename, O_RDONLY, 0)) < 0)
   {
      TRACE( Error, "Config() can't open configuration file " << config_filename);
      return false;
   }
 
   Config.Attach(fd);
   static const char *cvec[] = { "*** pfc plugin config:", 0 };
   Config.Capture(cvec);
 
   // Obtain OFS configurator for OSS plugin.
   XrdOfsConfigPI *ofsCfg = XrdOfsConfigPI::New(config_filename,&Config,&m_log,
                                                &XrdVERSIONINFOVAR(XrdOucGetCache));
   if (! ofsCfg) return false;
 
   TmpConfiguration tmpc;
 
   Configuration &CFG = m_configuration;
 
   // Adjust default parameters for client/serverless caching
   if (m_isClient)
   {
      m_configuration.m_bufferSize     = 128 * 1024; // same as normal.
      m_configuration.m_wqueue_blocks  = 8;
      m_configuration.m_wqueue_threads = 1;
   }
 
   // If network checksum processing is the default, indicate so.
   if (m_configuration.is_cschk_net()) m_env->Put("psx.CSNet", m_configuration.m_cs_ChkTLS ? "2" : "1");
 
   // Actual parsing of the config file.
   bool retval = true, aOK = true;
   char *var;
   while ((var = Config.GetMyFirstWord()))
   {
      if (! strcmp(var,"pfc.osslib"))
      {
         retval = ofsCfg->Parse(XrdOfsConfigPI::theOssLib);
      }
      else if (! strcmp(var,"pfc.cschk"))
      {
         retval = xcschk(Config);
      }
      else if (! strcmp(var,"pfc.decisionlib"))
      {
         retval = xdlib(Config);
      }
      else if (! strcmp(var,"pfc.purgelib"))
      {
         retval = xplib(Config);
      }
      else if (! strcmp(var,"pfc.trace"))
      {
         retval = xtrace(Config);
      }
      else if (! strcmp(var,"pfc.allow_xrdpfc_command"))
      {
         m_configuration.m_allow_xrdpfc_command = true;
      }
      else if (! strncmp(var,"pfc.", 4))
      {
         retval = ConfigParameters(std::string(var+4), Config, tmpc);
      }
 
      if ( ! retval)
      {
         TRACE(Error, "Config() error in parsing");
         aOK = false;
      }
   }
 
   Config.Close();
 
   // Load OSS plugin.
   auto orig_runmode = myEnv->Get("oss.runmode");
   myEnv->Put("oss.runmode", "pfc");
   if (m_configuration.is_cschk_cache())
   {
      char csi_conf[128];
      if (snprintf(csi_conf, 128, "space=%s nofill", m_configuration.m_meta_space.c_str()) < 128)
      {
         ofsCfg->Push(XrdOfsConfigPI::theOssLib, "libXrdOssCsi.so", csi_conf);
      } else {
         TRACE(Error, "Config() buffer too small for libXrdOssCsi params.");
         return false;
      }
   }
   if (ofsCfg->Load(XrdOfsConfigPI::theOssLib, myEnv))
   {
      ofsCfg->Plugin(m_oss);
   }
   else
   {
      TRACE(Error, "Config() Unable to create an OSS object");
      return false;
   }
   if (orig_runmode) myEnv->Put("oss.runmode", orig_runmode);
   else myEnv->Put("oss.runmode", "");
 
   // Test if OSS is operational, determine optional features.
   aOK &= test_oss_basics_and_features();
 
   // sets default value for disk usage
   XrdOssVSInfo sP;
   {
      if (m_configuration.m_meta_space != m_configuration.m_data_space &&
          m_oss->StatVS(&sP, m_configuration.m_meta_space.c_str(), 1) < 0)
      {
         m_log.Emsg("ConfigParameters()", "error obtaining stat info for meta space ", m_configuration.m_meta_space.c_str());
         return false;
      }
      if (m_configuration.m_meta_space != m_configuration.m_data_space && sP.Total < 10ll << 20)
      {
         m_log.Emsg("ConfigParameters()", "available data space is less than 10 MB (can be due to a mistake in oss.localroot directive) for space ",
                    m_configuration.m_meta_space.c_str());
                    return false;
      }
      if (m_oss->StatVS(&sP, m_configuration.m_data_space.c_str(), 1) < 0)
      {
         m_log.Emsg("ConfigParameters()", "error obtaining stat info for data space ", m_configuration.m_data_space.c_str());
         return false;
      }
      if (sP.Total < 10ll << 20)
      {
         m_log.Emsg("ConfigParameters()", "available data space is less than 10 MB (can be due to a mistake in oss.localroot directive) for space ",
                    m_configuration.m_data_space.c_str());
                    return false;
      }
 
      m_configuration.m_diskTotalSpace = sP.Total;
 
      if (cfg2bytes(tmpc.m_diskUsageLWM, m_configuration.m_diskUsageLWM, sP.Total, "lowWatermark") &&
          cfg2bytes(tmpc.m_diskUsageHWM, m_configuration.m_diskUsageHWM, sP.Total, "highWatermark"))
      {
         if (m_configuration.m_diskUsageLWM >= m_configuration.m_diskUsageHWM) {
            m_log.Emsg("ConfigParameters()", "pfc.diskusage should have lowWatermark < highWatermark.");
            aOK = false;
         }
      }
      else aOK = false;
 
      if ( ! tmpc.m_fileUsageMax.empty())
      {
        if (cfg2bytes(tmpc.m_fileUsageBaseline, m_configuration.m_fileUsageBaseline, sP.Total, "files baseline") &&
            cfg2bytes(tmpc.m_fileUsageNominal,  m_configuration.m_fileUsageNominal,  sP.Total, "files nominal")  &&
            cfg2bytes(tmpc.m_fileUsageMax,      m_configuration.m_fileUsageMax,      sP.Total, "files max"))
        {
          if (m_configuration.m_fileUsageBaseline >= m_configuration.m_fileUsageNominal ||
              m_configuration.m_fileUsageBaseline >= m_configuration.m_fileUsageMax     ||
              m_configuration.m_fileUsageNominal  >= m_configuration.m_fileUsageMax)
          {
            m_log.Emsg("ConfigParameters()", "pfc.diskusage files should have baseline < nominal < max.");
            aOK = false;
          }
 
 
         if (aOK && m_configuration.m_fileUsageMax >= m_configuration.m_diskUsageLWM)
         {
            m_log.Emsg("ConfigParameters()", "pfc.diskusage files values must be below lowWatermark");
            aOK = false;
         }
        }
        else aOK = false;
      }
   }
 
   // sets flush frequency
   if ( ! tmpc.m_flushRaw.empty())
   {
      if (::isalpha(*(tmpc.m_flushRaw.rbegin())))
      {
         if (XrdOuca2x::a2sz(m_log, "Error getting number of bytes written before flush",  tmpc.m_flushRaw.c_str(),
                             &m_configuration.m_flushCnt,
                             100 * m_configuration.m_bufferSize , 100000 * m_configuration.m_bufferSize))
         {
            return false;
         }
         m_configuration.m_flushCnt /= m_configuration.m_bufferSize;
      }
      else
      {
         if (XrdOuca2x::a2ll(m_log, "Error getting number of blocks written before flush", tmpc.m_flushRaw.c_str(),
                             &m_configuration.m_flushCnt, 100, 100000))
         {
            return false;
         }
      }
   }
 
   // get number of available RAM blocks after process configuration
   if (m_configuration.m_RamAbsAvailable == 0)
   {
      m_configuration.m_RamAbsAvailable = m_isClient ? 256ll * 1024 * 1024 : 1024ll * 1024 * 1024;
      char buff[1024];
      snprintf(buff, sizeof(buff), "RAM usage pfc.ram is not specified. Default value %s is used.", m_isClient ? "256m" : "1g");
      m_log.Say("Config info: ", buff);
   }
   // Setup number of standard-size blocks not released back to the system to 5% of total RAM.
   m_configuration.m_RamKeepStdBlocks = (m_configuration.m_RamAbsAvailable / m_configuration.m_bufferSize + 1) * 5 / 100;
 
   // Set tracing to debug if this is set in environment
   char* cenv = getenv("XRDDEBUG");
   if (cenv && ! strcmp(cenv,"1") && m_trace->What < 4) m_trace->What = 4;
 
   if (aOK)
   {
//                         000    001            010
      const char *csc[] = {"off", "cache nonet", "nocache net notls",
//                         011
                           "cache net notls",
//                         100    101            110
                           "off", "cache nonet", "nocache net tls",
//                         111
                           "cache net tls"};
      char uvk[32];
      if (m_configuration.m_cs_UVKeep < 0)
         strcpy(uvk, "lru");
      else
         sprintf(uvk, "%lld", (long long) m_configuration.m_cs_UVKeep);
      float ram_gb = (m_configuration.m_RamAbsAvailable) / float(1024*1024*1024);
 
      char urlcgi_blks[64] = "ignore", urlcgi_npref[32] = "ignore";
      if (CFG.m_cgi_blocksize_allowed)
         snprintf(urlcgi_blks, sizeof(urlcgi_blks), "%lldk %lldk",
                  CFG.m_cgi_min_bufferSize >> 10, CFG.m_cgi_max_bufferSize >> 10);
      if (CFG.m_cgi_prefetch_allowed)
         snprintf(urlcgi_npref, sizeof(urlcgi_npref), "%d %d",
                  CFG.m_cgi_min_prefetch_max_blocks, CFG.m_cgi_max_prefetch_max_blocks);
 
      char buff[8192];
      int  loff = 0;
      loff = snprintf(buff, sizeof(buff), "Config effective %s pfc configuration:\n"
                      "       pfc.cschk %s uvkeep %s\n"
                      "       pfc.blocksize %lldk\n"
                      "       pfc.prefetch %d\n"
                      "       pfc.urlcgi blocksize %s prefetch %s\n"
                      "       pfc.ram %.fg\n"
                      "       pfc.writequeue %d %d\n"
                      "       # Total available disk: %lld\n"
                      "       pfc.diskusage %lld %lld files %lld %lld %lld purgeinterval %d purgecoldfiles %d\n"
                      "       pfc.spaces %s %s\n"
                      "       pfc.trace %d\n"
                      "       pfc.flush %lld\n"
                      "       pfc.acchistorysize %d\n"
                      "       pfc.onlyIfCachedMinBytes %lld\n"
                      "       pfc.onlyIfCachedMinFrac %.2f\n",
                      config_filename,
                      csc[int(m_configuration.m_cs_Chk)], uvk,
                      m_configuration.m_bufferSize >> 10,
                      m_configuration.m_prefetch_max_blocks,
                      urlcgi_blks, urlcgi_npref,
                      ram_gb,
                      m_configuration.m_wqueue_blocks, m_configuration.m_wqueue_threads,
                      sP.Total,
                      m_configuration.m_diskUsageLWM, m_configuration.m_diskUsageHWM,
                      m_configuration.m_fileUsageBaseline, m_configuration.m_fileUsageNominal, m_configuration.m_fileUsageMax,
                      m_configuration.m_purgeInterval, m_configuration.m_purgeColdFilesAge,
                      m_configuration.m_data_space.c_str(),
                      m_configuration.m_meta_space.c_str(),
                      m_trace->What,
                      m_configuration.m_flushCnt,
                      m_configuration.m_accHistorySize,
                      m_configuration.m_onlyIfCachedMinSize,
                      m_configuration.m_onlyIfCachedMinFrac);
 
      if (m_configuration.is_dir_stat_reporting_on())
      {
         loff += snprintf(buff + loff, sizeof(buff) - loff,
                          "       pfc.dirstats interval %d maxdepth %d (internal: size_of_dirlist %d, size_of_globlist %d)\n",
                          m_configuration.m_dirStatsInterval, m_configuration.m_dirStatsStoreDepth,
                          (int) m_configuration.m_dirStatsDirs.size(), (int) m_configuration.m_dirStatsDirGlobs.size());
         loff += snprintf(buff + loff, sizeof(buff) - loff, "           dirlist:\n");
         for (std::set<std::string>::iterator i = m_configuration.m_dirStatsDirs.begin(); i != m_configuration.m_dirStatsDirs.end(); ++i)
            loff += snprintf(buff + loff, sizeof(buff) - loff, "               %s\n", i->c_str());
         loff += snprintf(buff + loff, sizeof(buff) - loff, "           globlist:\n");
         for (std::set<std::string>::iterator i = m_configuration.m_dirStatsDirGlobs.begin(); i != m_configuration.m_dirStatsDirGlobs.end(); ++i)
            loff += snprintf(buff + loff, sizeof(buff) - loff, "               %s/*\n", i->c_str());
      }
 
      if (m_configuration.m_hdfsmode)
      {
         loff += snprintf(buff + loff, sizeof(buff) - loff, "       pfc.hdfsmode hdfsbsize %lld\n", m_configuration.m_hdfsbsize);
      }
 
      if (m_configuration.m_username.empty())
      {
         char unameBuff[256];
         XrdOucUtils::UserName(getuid(), unameBuff, sizeof(unameBuff));
         m_configuration.m_username = unameBuff;
      }
      else
      {
         loff += snprintf(buff + loff, sizeof(buff) - loff, "       pfc.user %s\n", m_configuration.m_username.c_str());
      }
 
      m_log.Say(buff);
 
      m_env->Put("XRDPFC.SEGSIZE", std::to_string(m_configuration.m_bufferSize).c_str());
   }
 
   // Derived settings
   m_prefetch_enabled   = CFG.m_prefetch_max_blocks > 0 || CFG.m_cgi_max_prefetch_max_blocks > 0;
   Info::s_maxNumAccess = CFG.m_accHistorySize;
 
   m_gstream = (XrdXrootdGStream*) m_env->GetPtr("pfc.gStream*");
 
   m_log.Say("       pfc g-stream has", m_gstream ? "" : " NOT", " been configured via xrootd.monitor directive\n");
 
   // Create the ResourceMonitor and get it ready for starting the main thread function.
   if (aOK)
   {
      m_res_mon = new ResourceMonitor(*m_oss);
      m_res_mon->init_before_main();
   }
 
   m_log.Say("=====> Proxy file cache configuration parsing ", aOK ? "completed" : "failed");
 
   if (ofsCfg) delete ofsCfg;
 
   // XXXX-CKSUM Testing. To be removed after OssPgi is also merged and valildated.
   // Building of xrdpfc_print fails when this is enabled.
#ifdef XRDPFC_CKSUM_TEST
   {
      int xxx = m_configuration.m_cs_Chk;
 
      for (m_configuration.m_cs_Chk = CSChk_None; m_configuration.m_cs_Chk <= CSChk_Both; ++m_configuration.m_cs_Chk)
      {
         Info::TestCksumStuff();
      }
 
      m_configuration.m_cs_Chk = xxx;
   }
#endif
 
   return aOK;
}
 
//------------------------------------------------------------------------------
 
bool Cache::ConfigParameters(std::string part, XrdOucStream& config, TmpConfiguration &tmpc)
{
   struct ConfWordGetter
   {
      XrdOucStream &m_config;
      char         *m_last_word;
 
      ConfWordGetter(XrdOucStream& c) : m_config(c), m_last_word((char*)1) {}
 
      const char* GetWord() { if (HasLast()) m_last_word = m_config.GetWord(); return HasLast() ? m_last_word : ""; }
      bool        HasLast() { return (m_last_word != 0); }
   };
 
   ConfWordGetter cwg(config);
 
   Configuration &CFG = m_configuration;
 
   if ( part == "user" )
   {
      m_configuration.m_username = cwg.GetWord();
      if ( ! cwg.HasLast())
      {
         m_log.Emsg("Config", "Error: pfc.user requires a parameter.");
         return false;
      }
   }
   else if ( part == "diskusage" )
   {
      tmpc.m_diskUsageLWM = cwg.GetWord();
      tmpc.m_diskUsageHWM = cwg.GetWord();
 
      if (tmpc.m_diskUsageHWM.empty())
      {
         m_log.Emsg("Config", "Error: pfc.diskusage parameter requires at least two arguments.");
         return false;
      }
 
      const char *p = 0;
      while ((p = cwg.GetWord()) && cwg.HasLast())
      {
         if (strcmp(p, "files") == 0)
         {
            tmpc.m_fileUsageBaseline = cwg.GetWord();
            tmpc.m_fileUsageNominal  = cwg.GetWord();
            tmpc.m_fileUsageMax      = cwg.GetWord();
 
            if ( ! cwg.HasLast())
            {
               m_log.Emsg("Config", "Error: pfc.diskusage files directive requires three arguments.");
               return false;
            }
         }
         else if (strcmp(p, "sleep") == 0 || strcmp(p, "purgeinterval") == 0)
         {
            if (strcmp(p, "sleep") == 0) m_log.Emsg("Config", "warning sleep directive is deprecated in pfc.diskusage. Please use purgeinterval instead.");
 
            if (XrdOuca2x::a2tm(m_log, "Error getting purgeinterval", cwg.GetWord(), &m_configuration.m_purgeInterval, 60, 3600))
            {
               return false;
            }
         }
         else if (strcmp(p, "purgecoldfiles") == 0)
         {
            if (XrdOuca2x::a2tm(m_log, "Error getting purgecoldfiles age", cwg.GetWord(), &m_configuration.m_purgeColdFilesAge, 3600, 3600*24*360))
            {
               return false;
            }
            if (XrdOuca2x::a2i(m_log, "Error getting purgecoldfiles period", cwg.GetWord(), &m_configuration.m_purgeAgeBasedPeriod, 1, 1000))
            {
               return false;
            }
         }
         else
         {
            m_log.Emsg("Config", "Error: diskusage stanza contains unknown directive", p);
         }
      }
   }
   else if ( part == "acchistorysize" )
   {
      if ( XrdOuca2x::a2i(m_log, "Error getting access-history-size", cwg.GetWord(), &m_configuration.m_accHistorySize, 20, 200))
      {
         return false;
      }
   }
   else if ( part == "dirstats" )
   {
      const char *p = 0;
      while ((p = cwg.GetWord()) && cwg.HasLast())
      {
         if (strcmp(p, "interval") == 0)
         {
            int validIntervals[] = {60, 120, 300, 600, 900, 1200, 1800, 3600};
            int size = sizeof(validIntervals) / sizeof(int);
 
            if (XrdOuca2x::a2tm(m_log, "Error getting dirstsat interval", cwg.GetWord(),
                &m_configuration.m_dirStatsInterval, validIntervals[0], validIntervals[size - 1]))
            {
               return false;
            }
            bool match = false, round_down = false;
            for (int i = 0; i < size; i++) {
              if (validIntervals[i] == m_configuration.m_dirStatsInterval) {
                 match = true;
                 break;
              }
              if (i > 0 && m_configuration.m_dirStatsInterval < validIntervals[i]) {
                 m_configuration.m_dirStatsInterval = validIntervals[i - 1];
                 round_down = true;
                 break;
              }
            }
            if ( ! match && ! round_down) {
                m_log.Emsg("Config", "Error: dirstat interval parsing failed.");
               return false;
            }
            if (round_down) {
                m_log.Emsg("Config", "Info: dirstat interval was rounded down to the nearest valid value.");
            }
 
         }
         else if (strcmp(p, "maxdepth") == 0)
         {
            if (XrdOuca2x::a2i(m_log, "Error getting maxdepth value", cwg.GetWord(),
                               &m_configuration.m_dirStatsStoreDepth, 0, 16))
            {
               return false;
            }
         }
         else if (strcmp(p, "dir") == 0)
         {
            p = cwg.GetWord();
            if (p && p[0] == '/')
            {
               // XXX -- should we just store them as sets of PathTokenizer objects, not strings?
 
               char d[1024]; d[0] = 0;
               int  depth = 0;
               {  // Compress multiple slashes and "measure" depth
                  const char *pp = p;
                  char *pd = d;
                  *(pd++) = *(pp++);
                  while (*pp != 0)
                  {
                     if (*(pd - 1) == '/')
                     {
                        if (*pp == '/')
                        {
                           ++pp; continue;
                        }
                        ++depth;
                     }
                     *(pd++) = *(pp++);
                  }
                  *(pd--) = 0;
                  // remove trailing but but not leading /
                  if (*pd == '/' && pd != d) *pd = 0;
               }
               int ld = strlen(d);
               if (ld >= 2 && d[ld-1] == '*' && d[ld-2] == '/')
               {
                  d[ld-2] = 0;
                  ld     -= 2;
                  m_configuration.m_dirStatsDirGlobs.insert(d);
                  printf("Glob %s -> %s -- depth = %d\n", p, d, depth);
               }
               else
               {
                  m_configuration.m_dirStatsDirs.insert(d);
                  printf("Dir  %s -> %s -- depth = %d\n", p, d, depth);
               }
 
               m_configuration.m_dirStatsStoreDepth = std::max(m_configuration.m_dirStatsStoreDepth, depth);
            }
            else
            {
               m_log.Emsg("Config", "Error: dirstats dir parameter requires a directory argument starting with a '/'.");
               return false;
            }
         }
         else
         {
            m_log.Emsg("Config", "Error: dirstats stanza contains unknown directive '", p, "'");
            return false;
         }
      }
   }
   else if ( part == "blocksize" )
   {
      if ( ! blocksize_str2value("Config", cwg.GetWord(), CFG.m_bufferSize,
                                 CFG.s_min_bufferSize, CFG.s_max_bufferSize))
         return false;
   }
   else if ( part == "prefetch" || part == "nramprefetch" )
   {
      if (part == "nramprefetch")
      {
         m_log.Emsg("Config", "pfc.nramprefetch is deprecated, please use pfc.prefetch instead. Replacing the directive internally.");
      }
 
      if ( ! prefetch_str2value("Config", cwg.GetWord(), CFG.m_prefetch_max_blocks,
                                0, CFG.s_max_prefetch_max_blocks))
         return false;
   }
   else if ( part == "urlcgi" )
   {
      //  pfc.urlcgi [blocksize {ignore | min max}] [prefetch {ignore | min max}]
      const char *p = 0;
      while ((p = cwg.GetWord()) && cwg.HasLast())
      {
         if (strcmp(p, "blocksize") == 0)
         {
            std::string bmin = cwg.GetWord();
            if (bmin == "ignore")
               continue;
            std::string bmax = cwg.GetWord();
            if ( ! cwg.HasLast()) {
               m_log.Emsg("Config", "Error: pfc.urlcgi blocksize parameter requires two arguments.");
               return false;
            }
            if ( ! blocksize_str2value("Config::urlcgi", bmin.c_str(), CFG.m_cgi_min_bufferSize,
                                       CFG.s_min_bufferSize, CFG.s_max_bufferSize))
               return false;
            if ( ! blocksize_str2value("Config::urlcgi", bmax.c_str(), CFG.m_cgi_max_bufferSize,
                                       CFG.s_min_bufferSize, CFG.s_max_bufferSize))
               return false;
            if (CFG.m_cgi_min_bufferSize > CFG.m_cgi_max_bufferSize) {
               m_log.Emsg("Config", "Error: pfc.urlcgi blocksize second argument must be larger or equal to the first one.");
               return false;
            }
            CFG.m_cgi_blocksize_allowed = true;
         }
         else if (strcmp(p, "prefetch") == 0)
         {
            std::string bmin = cwg.GetWord();
            if (bmin == "ignore")
               continue;
            std::string bmax = cwg.GetWord();
            if ( ! cwg.HasLast()) {
               m_log.Emsg("Config", "Error: pfc.urlcgi blocksize parameter requires two arguments.");
               return false;
            }
            if ( ! prefetch_str2value("Config::urlcgi", bmin.c_str(), CFG.m_cgi_min_prefetch_max_blocks,
                                       0, CFG.s_max_prefetch_max_blocks))
               return false;
            if ( ! prefetch_str2value("Config::urlcgi", bmax.c_str(), CFG.m_cgi_max_prefetch_max_blocks,
                                      0, CFG.s_max_prefetch_max_blocks))
               return false;
            if (CFG.m_cgi_min_prefetch_max_blocks > CFG.m_cgi_max_prefetch_max_blocks) {
               m_log.Emsg("Config", "Error: pfc.urlcgi prefetch second argument must be larger or equal to the first one.");
               return false;
            }
            CFG.m_cgi_prefetch_allowed = true;
         }
         else
         {
            m_log.Emsg("Config", "Error: urlcgi stanza contains unknown directive '", p, "'");
            return false;
         }
      } // while get next pfc.urlcgi word
   }
   else if ( part == "nramread" )
   {
      m_log.Emsg("Config", "pfc.nramread is deprecated, please use pfc.ram instead. Ignoring this directive.");
      cwg.GetWord(); // Ignoring argument.
   }
   else if ( part == "ram" )
   {
      long long minRAM = m_isClient ? 256 * 1024 * 1024 : 1024 * 1024 * 1024;
      long long maxRAM = 256 * minRAM;
      if ( XrdOuca2x::a2sz(m_log, "get RAM available", cwg.GetWord(), &m_configuration.m_RamAbsAvailable, minRAM, maxRAM))
      {
         return false;
      }
   }
   else if ( part == "writequeue")
   {
      if (XrdOuca2x::a2i(m_log, "Error getting pfc.writequeue num-blocks", cwg.GetWord(), &m_configuration.m_wqueue_blocks, 1, 1024))
      {
         return false;
      }
      if (XrdOuca2x::a2i(m_log, "Error getting pfc.writequeue num-threads", cwg.GetWord(), &m_configuration.m_wqueue_threads, 1, 64))
      {
         return false;
      }
   }
   else if ( part == "spaces" )
   {
      m_configuration.m_data_space = cwg.GetWord();
      m_configuration.m_meta_space = cwg.GetWord();
      if ( ! cwg.HasLast())
      {
         m_log.Emsg("Config", "spacenames requires two parameters: <data-space> <metadata-space>.");
         return false;
      }
   }
   else if ( part == "hdfsmode" )
   {
      m_log.Emsg("Config", "pfc.hdfsmode is currently unsupported.");
      return false;
 
      m_configuration.m_hdfsmode = true;
 
      const char* params = cwg.GetWord();
      if (params)
      {
         if (! strncmp("hdfsbsize", params, 9))
         {
            long long minBlSize =  32 * 1024;
            long long maxBlSize = 128 * 1024 * 1024;
            if ( XrdOuca2x::a2sz(m_log, "Error getting file fragment size", cwg.GetWord(), &m_configuration.m_hdfsbsize, minBlSize, maxBlSize))
            {
               return false;
            }
         }
         else
         {
            m_log.Emsg("Config", "Error setting the fragment size parameter name");
            return false;
         }
      }
   }
   else if ( part == "flush" )
   {
      tmpc.m_flushRaw = cwg.GetWord();
      if ( ! cwg.HasLast())
      {
         m_log.Emsg("Config", "Error: pfc.flush requires a parameter.");
         return false;
      }
   }
   else if ( part == "onlyifcached" )
   {
      const char *p = 0;
      while ((p = cwg.GetWord()) && cwg.HasLast())
      {
         if (strcmp(p, "minsize") == 0)
         {
            std::string minBytes = cwg.GetWord();
            long long minBytesTop = 1024 * 1024 * 1024;
            if (::isalpha(*(minBytes.rbegin())))
            {
               if (XrdOuca2x::a2sz(m_log, "Error in parsing minsize value for onlyifcached parameter", minBytes.c_str(), &m_configuration.m_onlyIfCachedMinSize, 0, minBytesTop))
               {
                  return false;
               }
            }
            else
            {
               if (XrdOuca2x::a2ll(m_log, "Error in parsing numeric minsize value for onlyifcached parameter", minBytes.c_str(),&m_configuration.m_onlyIfCachedMinSize, 0, minBytesTop))
               {
                  return false;
               }
            }
         }
         if (strcmp(p, "minfrac") == 0)
         {
            std::string minFrac = cwg.GetWord();
            char *eP;
            errno = 0;
            double frac = strtod(minFrac.c_str(), &eP);
            if (errno || eP == minFrac.c_str())
            {
               m_log.Emsg("Config", "Error setting fraction for only-if-cached directive");
               return false;
            }
            m_configuration.m_onlyIfCachedMinFrac = frac;
         }
         else
         {
            m_log.Emsg("Config", "Error: onlyifcached stanza contains unknown directive", p);
         }
      }
   }
   else
   {
      m_log.Emsg("ConfigParameters() unmatched pfc parameter", part.c_str());
      return false;
   }
 
   return true;
}