Loading...
Searching...
No Matches
XrdPfcCommand.cc
Go to the documentation of this file.
//----------------------------------------------------------------------------------
// Copyright (c) 2014 by Board of Trustees of the Leland Stanford, Jr., University
// Author: Alja Mrak-Tadel
//----------------------------------------------------------------------------------
// XRootD is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// XRootD is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with XRootD. If not, see <http://www.gnu.org/licenses/>.
//----------------------------------------------------------------------------------
#include "XrdPfcInfo.hh"
#include "XrdPfc.hh"
#include "XrdPfcTrace.hh"
#include "XrdPfcPathParseTools.hh"
#include "XrdPfcResourceMonitor.hh"
#include "XrdOfs/XrdOfsConfigPI.hh"
#include "XrdOss/XrdOss.hh"
#include "XrdOuc/XrdOuca2x.hh"
#include "XrdOuc/XrdOucEnv.hh"
#include "XrdOuc/XrdOucStream.hh"
#include "XrdSys/XrdSysLogger.hh"
#include "XrdSys/XrdSysFallocate.hh"
#include <algorithm>
#include <cstring>
#include <iostream>
#include <fcntl.h>
#include <vector>
#include <sys/time.h>
namespace
{
const int MAX_ACCESSES = 20;
const long long ONE_MB = 1024ll * 1024;
const long long ONE_GB = 1024ll * 1024 * 1024;
}
using namespace XrdPfc;
//______________________________________________________________________________
void Cache::ExecuteCommandUrl(const std::string& command_url)
{
static const char *top_epfx = "ExecuteCommandUrl ";
SplitParser cp(command_url, "/");
std::string token = cp.get_token();
if (token != "xrdpfc_command")
{
TRACE(Error, top_epfx << "First token is NOT xrdpfc_command.");
return;
}
// Get the command
token = cp.get_token_as_string();
auto get_opt = [](SplitParser &sp) -> char {
const char *t = sp.get_token();
if (t)
return (t[0] == '-' && t[1] != 0) ? t[1] : 0;
else
return -1;
};
//================================================================
// create_file
//================================================================
if (token == "create_file")
{
static const char* err_prefix = "ExecuteCommandUrl: /xrdpfc_command/create_file: ";
static const char* usage =
"Usage: create_file/ [-h] [-s filesize] [-b blocksize] [-t access_time] [-d access_duration]/<path>\n"
" Creates a cache file with given parameters. Data in file is random.\n"
" Useful for cache purge testing.\n"
"Notes:\n"
" . If no options are needed one should still leave a space between / separators, ie., '/ /'\n"
" . Default filesize=1G, blocksize=<as configured>, access_time=-10, access_duration=10.\n"
" . -t and -d can be given multiple times to record several accesses.\n"
" . Negative arguments given to -t are interpreted as relative to now.\n";
const Configuration &conf = m_configuration;
token = cp.get_token_as_string();
TRACE(Debug, err_prefix << "Entered with argument string '" << token <<"'.");
if (token.empty()) {
TRACE(Error, err_prefix << "Options section must not be empty, a single space character is OK.");
return;
}
TRACE(Debug, err_prefix << "File path (reminder of URL) is '" << cp.get_reminder() <<"'.");
if ( ! cp.has_reminder()) {
TRACE(Error, err_prefix << "Path section must not be empty.");
return;
}
long long file_size = ONE_GB;
long long block_size = conf.m_bufferSize;
int access_time [MAX_ACCESSES];
int access_duration[MAX_ACCESSES];
int at_count = 0, ad_count = 0;
time_t time_now = time(0);
SplitParser ap(token, " ");
char theOpt;
while ((theOpt = get_opt(ap)) != (char) -1)
{
switch (theOpt)
{
case 'h': {
m_log.Say(err_prefix, " -- printing help, no action will be taken\n", usage);
return;
}
case 's': {
if (XrdOuca2x::a2sz(m_log, "Error getting filesize", ap.get_token(),
&file_size, 0ll, 32 * ONE_GB))
return;
break;
}
case 'b': {
if (XrdOuca2x::a2sz(m_log, "Error getting blocksize", ap.get_token(),
&block_size, 0ll, 64 * ONE_MB))
return;
break;
}
case 't': {
if (XrdOuca2x::a2i(m_log, "Error getting access time", ap.get_token(),
&access_time[at_count++], INT_MIN, INT_MAX))
return;
break;
}
case 'd': {
if (XrdOuca2x::a2i(m_log, "Error getting access duration", ap.get_token(),
&access_duration[ad_count++], 0, 24 * 3600))
return;
break;
}
default: {
TRACE(Error, err_prefix << "Unhandled command argument.");
return;
}
}
}
if (at_count < 1) access_time [at_count++] = time_now - 10;
if (ad_count < 1) access_duration[ad_count++] = 10;
if (at_count != ad_count)
{
TRACE(Error, err_prefix << "Options -t and -d must be given the same number of times.");
return;
}
std::string file_path (cp.get_reminder_with_delim());
std::string cinfo_path(file_path + Info::s_infoExtension);
TRACE(Debug, err_prefix << "Command arguments parsed successfully. Proceeding to create file " << file_path);
// Check if cinfo exists ... bail out if it does.
{
struct stat infoStat;
if (GetOss()->Stat(cinfo_path.c_str(), &infoStat) == XrdOssOK)
{
TRACE(Error, err_prefix << "cinfo file already exists for '" << file_path << "'. Refusing to overwrite.");
return;
}
}
TRACE(Debug, err_prefix << "Command arguments parsed successfully, proceeding to execution.");
{
const char *myUser = conf.m_username.c_str();
XrdOucEnv myEnv;
// Create the data file.
char size_str[32]; sprintf(size_str, "%lld", file_size);
myEnv.Put("oss.asize", size_str);
myEnv.Put("oss.cgroup", conf.m_data_space.c_str());
int cret;
if ((cret = GetOss()->Create(myUser, file_path.c_str(), 0600, myEnv, XRDOSS_mkpath)) != XrdOssOK)
{
TRACE(Error, err_prefix << "Create failed for data file " << file_path << ERRNO_AND_ERRSTR(-cret));
return;
}
XrdOssDF *myFile = GetOss()->newFile(myUser);
if ((cret = myFile->Open(file_path.c_str(), O_RDWR, 0600, myEnv)) != XrdOssOK)
{
TRACE(Error, err_prefix << "Open failed for data file " << file_path << ERRNO_AND_ERRSTR(-cret));
delete myFile;
return;
}
// Create the info file.
myEnv.Put("oss.asize", "64k"); // TODO: Calculate? Get it from configuration? Do not know length of access lists ...
myEnv.Put("oss.cgroup", conf.m_meta_space.c_str());
if ((cret = GetOss()->Create(myUser, cinfo_path.c_str(), 0600, myEnv, XRDOSS_mkpath)) != XrdOssOK)
{
TRACE(Error, err_prefix << "Create failed for info file " << cinfo_path << ERRNO_AND_ERRSTR(-cret));
myFile->Close(); delete myFile;
return;
}
XrdOssDF *myInfoFile = GetOss()->newFile(myUser);
if ((cret = myInfoFile->Open(cinfo_path.c_str(), O_RDWR, 0600, myEnv)) != XrdOssOK)
{
TRACE(Error, err_prefix << "Open failed for info file " << cinfo_path << ERRNO_AND_ERRSTR(-cret));
delete myInfoFile;
myFile->Close(); delete myFile;
return;
}
// Allocate space for the data file.
if ((cret = posix_fallocate(myFile->getFD(), 0, file_size)))
{
TRACE(Error, err_prefix << "posix_fallocate failed for data file " << file_path << ERRNO_AND_ERRSTR(cret));
}
// Fill up cinfo.
Info myInfo(m_trace, false);
myInfo.SetBufferSizeFileSizeAndCreationTime(block_size, file_size);
myInfo.SetAllBitsSynced();
for (int i = 0; i < at_count; ++i)
{
time_t att_time = access_time[i] >= 0 ? access_time[i] : time_now + access_time[i];
myInfo.WriteIOStatSingle(file_size, att_time, att_time + access_duration[i]);
}
myInfo.Write(myInfoFile, cinfo_path.c_str());
// Fake last modified time to the last access_time
{
time_t last_detach;
myInfo.GetLatestDetachTime(last_detach);
struct timespec acc_mod_time[2] = { {last_detach, UTIME_OMIT}, {last_detach, 0} };
futimens(myInfoFile->getFD(), acc_mod_time);
}
myInfoFile->Close(); delete myInfoFile;
myFile->Close(); delete myFile;
struct stat dstat;
GetOss()->Stat(file_path.c_str(), &dstat);
TRACE(Info, err_prefix << "Created file '" << file_path << "', size=" << (file_size>>20) << "MB, "
<< "st_blocks=" << dstat.st_blocks);
{
XrdSysCondVarHelper lock(&m_writeQ.condVar);
m_writeQ.writes_between_purges += file_size;
}
{
int token = m_res_mon->register_file_open(file_path, time_now, false);
XrdPfc::Stats stats;
stats.m_BytesWritten = file_size;
stats.m_StBlocksAdded = dstat.st_blocks;
m_res_mon->register_file_update_stats(token, stats);
m_res_mon->register_file_close(token, time(0), stats);
}
}
}
//================================================================
// remove_file
//================================================================
else if (token == "remove_file")
{
static const char* err_prefix = "ExecuteCommandUrl: /xrdpfc_command/remove_file: ";
static const char* usage =
"Usage: remove_file/ [-h] /<path>\n"
" Removes given file from the cache unless it is currently open.\n"
" Useful for removal of stale files or duplicate files in a caching cluster.\n"
"Notes:\n"
" . If no options are needed one should still leave a space between / separators, ie., '/ /'\n";
token = cp.get_token_as_string();
TRACE(Debug, err_prefix << "Entered with argument string '" << token <<"'.");
if (token.empty()) {
TRACE(Error, err_prefix << "Options section must not be empty, a single space character is OK.");
return;
}
TRACE(Debug, err_prefix << "File path (reminder of URL) is '" << cp.get_reminder() <<"'.");
if ( ! cp.has_reminder()) {
TRACE(Error, err_prefix << "Path section must not be empty.");
return;
}
SplitParser ap(token, " ");
char theOpt;
while ((theOpt = get_opt(ap)) != (char) -1)
{
switch (theOpt)
{
case 'h': {
m_log.Say(err_prefix, " -- printing help, no action will be taken\n", usage);
return;
}
default: {
TRACE(Error, err_prefix << "Unhandled command argument.");
return;
}
}
}
std::string f_name(cp.get_reminder_with_delim());
TRACE(Debug, err_prefix << "file argument '" << f_name << "'.");
int ret = UnlinkFile(f_name, true);
TRACE(Info, err_prefix << "returned with status " << ret);
}
//================================================================
// unknown command
//================================================================
else
{
TRACE(Error, top_epfx << "Unknown or empty command '" << token << "'");
}
}
//==============================================================================
// Example python script to use /xrdpfc_command/
//==============================================================================
/*
from XRootD import client
from XRootD.client.flags import OpenFlags
import sys
import time
#-------------------------------------------------------------------------------
port = int( sys.argv[1] );
g_srv = "root://localhost:%d/" % port
g_com = "/xrdpfc_command/create_file/"
g_dir = "/store/user/matevz/"
#-------------------------------------------------------------------------------
def xxsend(args, file) :
url = g_srv + g_com + args + g_dir + file
print "Opening ", url
with client.File() as f:
status, response = f.open(url, OpenFlags.READ)
print '%r' % status
print '%r' % response
#-------------------------------------------------------------------------------
pfx1 = "AAAA"
pfx2 = "BBBB"
for i in range(1, 1024 + 1):
atime = -10000 + i
xxsend("-s 4g -t %d -d 10" % atime,
"%s-%04d" % (pfx1, i))
time.sleep(0.01)
for i in range(1, 512 + 1):
atime = -5000 + i
xxsend("-s 4g -t %d -d 10" % atime,
"%s-%04d" % (pfx2, i))
time.sleep(0.01)
*/
Stat
struct stat Stat
Definition XrdCks.cc:49
usage
void usage()
Definition XrdClReplay.cc:1072
XrdOfsConfigPI.hh
XrdOss.hh
XrdOssOK
#define XrdOssOK
Definition XrdOss.hh:50
XRDOSS_mkpath
#define XRDOSS_mkpath
Definition XrdOss.hh:466
XrdOucEnv.hh
XrdOucStream.hh
XrdOuca2x.hh
XrdPfcInfo.hh
XrdPfcPathParseTools.hh
XrdPfcResourceMonitor.hh
XrdPfcTrace.hh
ERRNO_AND_ERRSTR
#define ERRNO_AND_ERRSTR(err_code)
Definition XrdPfcTrace.hh:46
XrdPfc.hh
stat
#define stat(a, b)
Definition XrdPosix.hh:101
Debug
bool Debug
Definition XrdSecgsiProxy.cc:95
Create
bool Create
Definition XrdSecpwdSrvAdmin.cc:176
XrdSysFallocate.hh
XrdSysLogger.hh
Error
@ Error
Definition XrdTlsNotaryUtils.hh:42
TRACE
#define TRACE(act, x)
Definition XrdTrace.hh:63
XrdOssDF
Definition XrdOss.hh:63
XrdOssDF::Close
virtual int Close(long long *retsz=0)=0
XrdOssDF::getFD
virtual int getFD()
Definition XrdOss.hh:426
XrdOssDF::Open
virtual int Open(const char *path, int Oflag, mode_t Mode, XrdOucEnv &env)
Definition XrdOss.hh:200
XrdOss::newFile
virtual XrdOssDF * newFile(const char *tident)=0
XrdOss::Stat
virtual int Stat(const char *path, struct stat *buff, int opts=0, XrdOucEnv *envP=0)=0
XrdOucEnv
Definition XrdOucEnv.hh:42
XrdOucEnv::Put
void Put(const char *varname, const char *value)
Definition XrdOucEnv.hh:85
XrdOuca2x::a2i
static int a2i(XrdSysError &, const char *emsg, const char *item, int *val, int minv=-1, int maxv=-1)
Definition XrdOuca2x.cc:45
XrdOuca2x::a2sz
static int a2sz(XrdSysError &, const char *emsg, const char *item, long long *val, long long minv=-1, long long maxv=-1)
Definition XrdOuca2x.cc:257
XrdPfc::Cache::ExecuteCommandUrl
void ExecuteCommandUrl(const std::string &command_url)
Definition XrdPfcCommand.cc:51
XrdPfc::Cache::UnlinkFile
int UnlinkFile(const std::string &f_name, bool fail_if_open)
Remove cinfo and data files from cache.
Definition XrdPfc.cc:1188
XrdPfc::Cache::GetOss
XrdOss * GetOss() const
Definition XrdPfc.hh:280
XrdPfc::Info
Status of cached file. Can be read from and written into a binary file.
Definition XrdPfcInfo.hh:41
XrdPfc::Info::s_infoExtension
static const char * s_infoExtension
Definition XrdPfcInfo.hh:309
XrdPfc::Info::WriteIOStatSingle
void WriteIOStatSingle(long long bytes_disk)
Write single open/close time for given bytes read from disk.
Definition XrdPfcInfo.cc:444
XrdPfc::Info::GetLatestDetachTime
bool GetLatestDetachTime(time_t &t) const
Get latest detach time.
Definition XrdPfcInfo.cc:470
XrdPfc::Info::Write
bool Write(XrdOssDF *fp, const char *dname, const char *fname=0)
Definition XrdPfcInfo.cc:266
XrdPfc::Info::SetAllBitsSynced
void SetAllBitsSynced()
Mark all blocks as synced to disk.
Definition XrdPfcInfo.cc:144
XrdPfc::Info::SetBufferSizeFileSizeAndCreationTime
void SetBufferSizeFileSizeAndCreationTime(long long bs, long long fs)
Definition XrdPfcInfo.cc:161
XrdPfc::Stats
Statistics of cache utilisation by a File object.
Definition XrdPfcStats.hh:35
XrdPfc::Stats::m_StBlocksAdded
long long m_StBlocksAdded
number of 512-byte blocks the file has grown by
Definition XrdPfcStats.hh:43
XrdPfc::Stats::m_BytesWritten
long long m_BytesWritten
number of bytes written to disk
Definition XrdPfcStats.hh:42
XrdSysCondVarHelper
Definition XrdSysPthread.hh:129
XrdPfc
Definition XrdPfc.hh:41
XrdPfc::Configuration
Contains parameters configurable from the xrootd config file.
Definition XrdPfc.hh:64
XrdPfc::Configuration::m_data_space
std::string m_data_space
oss space for data files
Definition XrdPfc.hh:88
XrdPfc::Configuration::m_bufferSize
long long m_bufferSize
cache block size, default 128 kB
Definition XrdPfc.hh:107
XrdPfc::Configuration::m_meta_space
std::string m_meta_space
oss space for metadata files (cinfo)
Definition XrdPfc.hh:89
XrdPfc::Configuration::m_username
std::string m_username
username passed to oss plugin
Definition XrdPfc.hh:87
XrdPfc::SplitParser
Definition XrdPfcPathParseTools.hh:13
XrdPfc::SplitParser::get_reminder_with_delim
const char * get_reminder_with_delim()
Definition XrdPfcPathParseTools.hh:46
XrdPfc::SplitParser::get_token_as_string
std::string get_token_as_string()
Definition XrdPfcPathParseTools.hh:40
XrdPfc::SplitParser::get_reminder
const char * get_reminder()
Definition XrdPfcPathParseTools.hh:52
XrdPfc::SplitParser::get_token
const char * get_token()
Definition XrdPfcPathParseTools.hh:25
XrdPfc::SplitParser::has_reminder
bool has_reminder()
Definition XrdPfcPathParseTools.hh:57