XrdFileCache.hh

Go to the documentation of this file.

#ifndef __XRDFILECACHE_CACHE_HH__
#define __XRDFILECACHE_CACHE_HH__
//----------------------------------------------------------------------------------
// Copyright (c) 2014 by Board of Trustees of the Leland Stanford, Jr., University
// Author: Alja Mrak-Tadel, Matevz Tadel, Brian Bockelman
//----------------------------------------------------------------------------------
// 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 emacs 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 <string>
#include <list>
#include <set>
 
#include "Xrd/XrdScheduler.hh"
#include "XrdVersion.hh"
#include "XrdSys/XrdSysPthread.hh"
#include "XrdOuc/XrdOucCache2.hh"
#include "XrdOuc/XrdOucCallBack.hh"
#include "XrdCl/XrdClDefaultEnv.hh"
#include "XrdFileCacheFile.hh"
#include "XrdFileCacheDecision.hh"
 
class XrdOucStream;
class XrdSysError;
class XrdSysTrace;
 
namespace XrdCl {
class Log;
}
namespace XrdFileCache {
class File;
class IO;
}
 
 
namespace XrdFileCache
{
//----------------------------------------------------------------------------
//----------------------------------------------------------------------------
struct 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_purgeColdFilesPeriod(-1),
      m_bufferSize(1024*1024),
      m_RamAbsAvailable(0),
      m_NRamBuffers(-1),
      m_wqueue_blocks(16),
      m_wqueue_threads(4),
      m_prefetch_max_blocks(10),
      m_hdfsbsize(128*1024*1024),
      m_flushCnt(2000)
   {}
 
   bool are_file_usage_limits_set()    const { return m_fileUsageMax > 0; }
   bool is_age_based_purge_in_effect() const { return m_purgeColdFilesAge > 0; }
   bool is_purge_plugin_set_up()       const { return false; }
 
   void calculate_fractional_usages(long long du, long long fu, double &frac_du, double &frac_fu);
 
   bool m_hdfsmode;                     
   bool m_allow_xrdpfc_command;         
 
   std::string m_username;              
   std::string m_data_space;            
   std::string m_meta_space;            
 
   long long m_diskTotalSpace;          
   long long m_diskUsageLWM;            
   long long m_diskUsageHWM;            
   long long m_fileUsageBaseline;       
   long long m_fileUsageNominal;        
   long long m_fileUsageMax;            
   int       m_purgeInterval;           
   int       m_purgeColdFilesAge;       
   int       m_purgeColdFilesPeriod;    
 
   long long m_bufferSize;              
   long long m_RamAbsAvailable;         
   int       m_NRamBuffers;             
   int       m_wqueue_blocks;           
   int       m_wqueue_threads;          
   int       m_prefetch_max_blocks;     
 
   long long m_hdfsbsize;               
   long long m_flushCnt;                
};
 
struct TmpConfiguration
{
   std::string m_diskUsageLWM;
   std::string m_diskUsageHWM;
   std::string m_fileUsageBaseline;
   std::string m_fileUsageNominal;
   std::string m_fileUsageMax;
   std::string m_flushRaw;
 
   TmpConfiguration() :
      m_diskUsageLWM("0.90"), m_diskUsageHWM("0.95"),
      m_flushRaw("")
   {}
};
 
//----------------------------------------------------------------------------
//----------------------------------------------------------------------------
class Cache : public XrdOucCache2
{
public:
   //---------------------------------------------------------------------
   //---------------------------------------------------------------------
   Cache(XrdSysLogger *logger);
 
   //---------------------------------------------------------------------
   //---------------------------------------------------------------------
   using XrdOucCache2::Attach;
 
   virtual XrdOucCacheIO2 *Attach(XrdOucCacheIO2 *, int Options = 0);
 
   //---------------------------------------------------------------------
   //---------------------------------------------------------------------
   virtual int isAttached();
 
   //---------------------------------------------------------------------
   // Virtual function of XrdOucCache2. Used to pass environmental info.
   virtual void EnvInfo(XrdOucEnv &theEnv);
 
   //---------------------------------------------------------------------
   // Virtual function of XrdOucCache2. Used for redirection to a local
   // file on a distributed FS.
   virtual int LocalFilePath(const char *url, char *buff=0, int blen=0,
                             LFP_Reason why=ForAccess, bool forall=false);
 
   //---------------------------------------------------------------------
   // Virtual function of XrdOucCache2. Used for deferred open.
   virtual int  Prepare(const char *url, int oflags, mode_t mode);
 
   // virtual function of XrdOucCache2.
   virtual int  Stat(const char *url, struct stat &sbuff);
 
   // virtual function of XrdOucCache.
   virtual int  Unlink(const char *url);
 
   //--------------------------------------------------------------------
   //--------------------------------------------------------------------
   bool Decide(XrdOucCacheIO*);
 
   //------------------------------------------------------------------------
   //------------------------------------------------------------------------
   const Configuration& RefConfiguration() const { return m_configuration; }
 
   //---------------------------------------------------------------------
   //---------------------------------------------------------------------
   bool Config(const char *config_filename, const char *parameters);
 
   //---------------------------------------------------------------------
   //---------------------------------------------------------------------
   static Cache &CreateInstance(XrdSysLogger *logger);
 
  //---------------------------------------------------------------------
   //---------------------------------------------------------------------
   static Cache &GetInstance();
 
   //---------------------------------------------------------------------
   //---------------------------------------------------------------------
   static bool VCheck(XrdVersionInfo &urVersion) { return true; }
 
   //---------------------------------------------------------------------
   //---------------------------------------------------------------------
   void Purge();
 
   //---------------------------------------------------------------------
   //---------------------------------------------------------------------
   int  UnlinkUnlessOpen(const std::string& f_name);
 
   //---------------------------------------------------------------------
   //---------------------------------------------------------------------
   void AddWriteTask(Block* b, bool from_read);
 
   //---------------------------------------------------------------------
   //---------------------------------------------------------------------
   void RemoveWriteQEntriesFor(File *f);
 
   //---------------------------------------------------------------------
   //---------------------------------------------------------------------
   void ProcessWriteTasks();
 
   bool RequestRAMBlock();
 
   void RAMBlockReleased();
 
   void RegisterPrefetchFile(File*);
   void DeRegisterPrefetchFile(File*);
 
   File* GetNextFileToPrefetch();
 
   void Prefetch();
 
   XrdOss* GetOss() const { return m_output_fs; }
 
   bool IsFileActiveOrPurgeProtected(const std::string&);
   
   File* GetFile(const std::string&, IO*, long long off = 0, long long filesize = 0);
 
   void  ReleaseFile(File*, IO*);
 
   void ScheduleFileSync(File* f) { schedule_file_sync(f, false, false); }
 
   void FileSyncDone(File*, bool high_debug);
   
   XrdSysError* GetLog()   { return &m_log;  }
   XrdSysTrace* GetTrace() { return m_trace; }
 
   void ExecuteCommandUrl(const std::string& command_url);
 
private:
   bool ConfigParameters(std::string, XrdOucStream&, TmpConfiguration &tmpc);
   bool ConfigXeq(char *, XrdOucStream &);
   bool xdlib(XrdOucStream &);
   bool xtrace(XrdOucStream &);
 
   bool cfg2bytes(const std::string &str, long long &store, long long totalSpace, const char *name);
 
   int  UnlinkCommon(const std::string& f_name, bool fail_if_open);
 
   static Cache        *m_factory;      
   static XrdScheduler *schedP;
 
   XrdSysError       m_log;             
   XrdSysTrace      *m_trace;
   const char       *m_traceID;
 
   XrdOucCacheStats  m_stats;           
   XrdOss           *m_output_fs;       
 
   std::vector<XrdFileCache::Decision*> m_decisionpoints;       
 
   std::map<std::string, long long> m_filesInQueue;
 
   Configuration m_configuration;           
 
   XrdSysCondVar m_prefetch_condVar;        
   bool          m_prefetch_enabled;        
 
   XrdSysMutex m_RAMblock_mutex;            
   int         m_RAMblocks_used;
   bool        m_isClient;                  
 
   struct WriteQ
   {
      WriteQ() : condVar(0), writes_between_purges(0), size(0) {}
 
      XrdSysCondVar     condVar;      
      std::list<Block*> queue;        
      long long         writes_between_purges; 
      int               size;         
   };
 
   WriteQ m_writeQ;
 
   // active map, purge delay set
   typedef std::map<std::string, File*> ActiveMap_t;
   typedef ActiveMap_t::iterator        ActiveMap_i;
   typedef std::set<std::string>        FNameSet_t;
 
   ActiveMap_t   m_active;
   FNameSet_t    m_purge_delay_set;
   bool          m_in_purge;
   XrdSysCondVar m_active_cond;
 
   void inc_ref_cnt(File*, bool lock, bool high_debug);
   void dec_ref_cnt(File*, bool high_debug);
 
   void schedule_file_sync(File*, bool ref_cnt_already_set, bool high_debug);
 
   // prefetching
   typedef std::vector<File*>  PrefetchList;
   PrefetchList m_prefetchList;
};
 
}
 
#endif