XrdPfcFile.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, Matevz 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 "XrdPfcFile.hh"
#include "XrdPfc.hh"
#include "XrdPfcResourceMonitor.hh"
#include "XrdPfcIO.hh"
#include "XrdPfcTrace.hh"
 
#include "XProtocol/XProtocol.hh"
#include "XrdSys/XrdSysTimer.hh"
#include "XrdOss/XrdOss.hh"
#include "XrdOuc/XrdOucEnv.hh"
#include "XrdSfs/XrdSfsInterface.hh"
 
#include "XrdCl/XrdClURL.hh"
 
#include <cassert>
#include <cstdio>
#include <sstream>
#include <unordered_map>
 
#include <fcntl.h>
 
using namespace XrdPfc;
 
namespace
{
 
const int BLOCK_WRITE_MAX_ATTEMPTS = 4;
 
Cache* cache() { return &Cache::GetInstance(); }
 
}
 
const char *File::m_traceID = "File";
 
//------------------------------------------------------------------------------
 
File::File(const std::string& path, long long iOffset, long long iFileSize) :
   m_ref_cnt(0),
   m_data_file(0),
   m_info_file(0),
   m_cfi(Cache::TheOne().GetTrace(), Cache::TheOne().is_prefetch_enabled()),
   m_filename(path),
   m_offset(iOffset),
   m_file_size(iFileSize),
   m_current_io(m_io_set.end()),
   m_ios_in_detach(0),
   m_non_flushed_cnt(0),
   m_in_sync(false),
   m_detach_time_logged(false),
   m_in_shutdown(false),
   m_state_cond(0),
   m_block_size(0),
   m_num_blocks(0),
   m_resmon_token(-1),
   m_prefetch_state(kOff),
   m_prefetch_bytes(0),
   m_prefetch_read_cnt(0),
   m_prefetch_hit_cnt(0),
   m_prefetch_score(0)
{}
 
File::~File()
{
   TRACEF(Debug, "~File() for ");
}
 
void File::Close()
{
   // Close is called while nullptr is put into Cache::m_active map, see Cache::dec_ref_count(File*).
   // A stat is called after close to re-check that m_stat_blocks have been reported correctly
   // to the resource-monitor. Note that the reporting is already clamped down to m_file_size
   // in report_and_merge_delta_stats() below.
   //
   // XFS can pre-allocate significant amount of blocks (1 GB at 1GB mark, 4 GB above 4GB) and those
   // get reported in as stat.st_blocks.
   // The reported number is correct in a stat immediately following a close.
   // If one starts off by writing the last byte of the file, this pre-allocation does not get
   // triggered up to that point. But comes back with a vengeance right after.
   //
   // To be determined if other FSes do something similar (Ceph, ZFS, ...). Ext4 doesn't.
 
   if (m_info_file)
   {
      TRACEF(Debug, "Close() closing info-file ");
      m_info_file->Close();
      delete m_info_file;
      m_info_file = nullptr;
   }
 
   if (m_data_file)
   {
      TRACEF(Debug, "Close() closing data-file ");
      m_data_file->Close();
      delete m_data_file;
      m_data_file = nullptr;
   }
 
   if (m_resmon_token >= 0)
   {
      // Last update of file stats has been sent from the final Sync unless we are in_shutdown --
      // but in this case the file will get unlinked by the cache and reported as purge event.
      // We check if the reported st_blocks so far is correct.
      if (m_stats.m_BytesWritten > 0 && ! m_in_shutdown) {
         struct stat s;
         int sr = Cache::GetInstance().GetOss()->Stat(m_filename.c_str(), &s);
         if (sr == 0 && s.st_blocks != m_st_blocks) {
            Stats stats;
            stats.m_StBlocksAdded = s.st_blocks - m_st_blocks;
            m_st_blocks = s.st_blocks;
            Cache::ResMon().register_file_update_stats(m_resmon_token, stats);
         }
      }
 
      Cache::ResMon().register_file_close(m_resmon_token, time(0), m_stats);
   }
 
   TRACEF(Debug, "Close() finished, prefetch score = " <<  m_prefetch_score);
}
 
//------------------------------------------------------------------------------
 
File* File::FileOpen(const std::string &path, long long offset, long long fileSize, XrdOucCacheIO* inputIO)
{
   File *file = new File(path, offset, fileSize);
   if ( ! file->Open(inputIO))
   {
      delete file;
      file = 0;
   }
   return file;
}
 
//------------------------------------------------------------------------------
 
long long File::initiate_emergency_shutdown()
{
   // Called from Cache::Unlink() when the file is currently open.
   // Cache::Unlink is also called on FSync error and when wrong number of bytes
   // is received from a remote read.
   //
   // From this point onward the file will not be written to, cinfo file will
   // not be updated, and all new read requests will return -ENOENT.
   //
   // File's entry in the Cache's active map is set to nullptr and will be
   // removed from there shortly, in any case, well before this File object
   // shuts down. Cache::Unlink() also reports the appropriate purge event.
 
   XrdSysCondVarHelper _lck(m_state_cond);
 
   m_in_shutdown = true;
 
   if (m_prefetch_state != kStopped && m_prefetch_state != kComplete)
   {
      m_prefetch_state = kStopped;
      cache()->DeRegisterPrefetchFile(this);
   }
 
   report_and_merge_delta_stats();
 
   return m_st_blocks;
}
 
//------------------------------------------------------------------------------
 
void File::check_delta_stats()
{
   // Called under m_state_cond lock.
   // BytesWritten indirectly trigger an unconditional merge through periodic Sync().
   if (m_delta_stats.BytesReadAndWritten() >= m_resmon_report_threshold && ! m_in_shutdown)
      report_and_merge_delta_stats();
}
 
void File::report_and_merge_delta_stats()
{
   // Called under m_state_cond lock.
   struct stat s;
   m_data_file->Fstat(&s);
   // Do not report st_blocks beyond 4kB round-up over m_file_size. Some FSs report
   // aggressive pre-allocation in this field (XFS, 4GB).
   long long max_st_blocks_to_report = (m_file_size & 0xfff) ? ((m_file_size >> 12) + 1) << 3
                                                             :   m_file_size >> 9;
   long long st_blocks_to_report = std::min((long long) s.st_blocks, max_st_blocks_to_report);
   m_delta_stats.m_StBlocksAdded = st_blocks_to_report - m_st_blocks;
   m_st_blocks = st_blocks_to_report;
   Cache::ResMon().register_file_update_stats(m_resmon_token, m_delta_stats);
   m_stats.AddUp(m_delta_stats);
   m_delta_stats.Reset();
}
 
//------------------------------------------------------------------------------
 
void File::BlockRemovedFromWriteQ(Block* b)
{
   TRACEF(Dump, "BlockRemovedFromWriteQ() block = " << (void*) b << " idx= " << b->m_offset/m_block_size);
 
   XrdSysCondVarHelper _lck(m_state_cond);
   dec_ref_count(b);
}
 
void File::BlocksRemovedFromWriteQ(std::list<Block*>& blocks)
{
   TRACEF(Dump, "BlocksRemovedFromWriteQ() n_blocks = " << blocks.size());
 
   XrdSysCondVarHelper _lck(m_state_cond);
 
   for (std::list<Block*>::iterator i = blocks.begin(); i != blocks.end(); ++i)
   {
      dec_ref_count(*i);
   }
}
 
//------------------------------------------------------------------------------
 
void File::ioUpdated(IO *io)
{
   std::string loc(io->GetLocation());
   XrdSysCondVarHelper _lck(m_state_cond);
   insert_remote_location(loc);
}
 
//------------------------------------------------------------------------------
 
bool File::ioActive(IO *io)
{
   // Returns true if delay is needed.
 
   TRACEF(Debug, "ioActive start for io " << io);
 
   std::string loc(io->GetLocation());
 
   {
      XrdSysCondVarHelper _lck(m_state_cond);
 
      IoSet_i mi = m_io_set.find(io);
 
      if (mi != m_io_set.end())
      {
         unsigned int n_active_reads = io->m_active_read_reqs;
 
         TRACE(Info, "ioActive for io " << io <<
                ", active_reads "       << n_active_reads <<
                ", active_prefetches "  << io->m_active_prefetches <<
                ", allow_prefetching "  << io->m_allow_prefetching <<
                ", ios_in_detach "      << m_ios_in_detach);
         TRACEF(Info,
                "\tio_map.size() "      << m_io_set.size() <<
                ", block_map.size() "   << m_block_map.size() << ", file");
 
         insert_remote_location(loc);
 
         io->m_allow_prefetching = false;
         io->m_in_detach = true;
 
         // Check if any IO is still available for prfetching. If not, stop it.
         if (m_prefetch_state == kOn || m_prefetch_state == kHold)
         {
            if ( ! select_current_io_or_disable_prefetching(false) )
            {
               TRACEF(Debug, "ioActive stopping prefetching after io " << io << " retreat.");
            }
         }
 
         // On last IO, consider write queue blocks. Note, this also contains
         // blocks being prefetched.
 
         bool io_active_result;
 
         if (n_active_reads > 0)
         {
            io_active_result = true;
         }
         else if (m_io_set.size() - m_ios_in_detach == 1)
         {
            io_active_result = ! m_block_map.empty();
         }
         else
         {
            io_active_result = io->m_active_prefetches > 0;
         }
 
         if ( ! io_active_result)
         {
            ++m_ios_in_detach;
         }
 
         TRACEF(Info, "ioActive for io " << io << " returning " << io_active_result << ", file");
 
         return io_active_result;
      }
      else
      {
         TRACEF(Error, "ioActive io " << io <<" not found in IoSet. This should not happen.");
         return false;
      }
   }
}
 
//------------------------------------------------------------------------------
 
void File::RequestSyncOfDetachStats()
{
   XrdSysCondVarHelper _lck(m_state_cond);
   m_detach_time_logged = false;
}
 
bool File::FinalizeSyncBeforeExit()
{
   // Returns true if sync is required.
   // This method is called after corresponding IO is detached from PosixCache.
 
   XrdSysCondVarHelper _lck(m_state_cond);
   if ( ! m_in_shutdown)
   {
     if ( ! m_writes_during_sync.empty() || m_non_flushed_cnt > 0 || ! m_detach_time_logged)
     {
       report_and_merge_delta_stats();
       m_cfi.WriteIOStatDetach(m_stats);
       m_detach_time_logged = true;
       m_in_sync            = true;
       TRACEF(Debug, "FinalizeSyncBeforeExit requesting sync to write detach stats");
       return true;
     }
   }
   TRACEF(Debug, "FinalizeSyncBeforeExit sync not required");
   return false;
}
 
//------------------------------------------------------------------------------
 
void File::AddIO(IO *io)
{
   // Called from Cache::GetFile() when a new IO asks for the file.
 
   TRACEF(Debug, "AddIO() io = " << (void*)io);
 
   time_t      now = time(0);
   std::string loc(io->GetLocation());
 
   m_state_cond.Lock();
 
   IoSet_i mi = m_io_set.find(io);
 
   if (mi == m_io_set.end())
   {
      m_io_set.insert(io);
      io->m_attach_time = now;
      m_delta_stats.IoAttach();
 
      insert_remote_location(loc);
 
      if (m_prefetch_state == kStopped)
      {
         m_prefetch_state = kOn;
         cache()->RegisterPrefetchFile(this);
      }
   }
   else
   {
      TRACEF(Error, "AddIO() io = " << (void*)io << " already registered.");
   }
 
   m_state_cond.UnLock();
}
 
//------------------------------------------------------------------------------
 
void File::RemoveIO(IO *io)
{
   // Called from Cache::ReleaseFile.
 
   TRACEF(Debug, "RemoveIO() io = " << (void*)io);
 
   time_t now = time(0);
 
   m_state_cond.Lock();
 
   IoSet_i mi = m_io_set.find(io);
 
   if (mi != m_io_set.end())
   {
      if (mi == m_current_io)
      {
         ++m_current_io;
      }
 
      m_delta_stats.IoDetach(now - io->m_attach_time);
      m_io_set.erase(mi);
      --m_ios_in_detach;
 
      if (m_io_set.empty() && m_prefetch_state != kStopped && m_prefetch_state != kComplete)
      {
         TRACEF(Error, "RemoveIO() io = " << (void*)io << " Prefetching is not stopped/complete -- it should be by now.");
         m_prefetch_state = kStopped;
         cache()->DeRegisterPrefetchFile(this);
      }
   }
   else
   {
      TRACEF(Error, "RemoveIO() io = " << (void*)io << " is NOT registered.");
   }
 
   m_state_cond.UnLock();
}
 
//------------------------------------------------------------------------------
 
bool File::Open(XrdOucCacheIO* inputIO)
{
   // Sets errno accordingly.
 
   static const char *tpfx = "Open() ";
 
   TRACEF(Dump, tpfx << "entered");
 
   // Before touching anything, check with ResourceMonitor if a scan is in progress.
   // This function will wait internally if needed until it is safe to proceed.
   Cache::ResMon().CrossCheckIfScanIsInProgress(m_filename, m_state_cond);
 
   const Configuration &conf = Cache::GetInstance().RefConfiguration();
 
   XrdOss     &myOss  = * Cache::GetInstance().GetOss();
   const char *myUser =   conf.m_username.c_str();
   XrdOucEnv   myEnv;
   struct stat data_stat, info_stat;
 
   std::string ifn = m_filename + Info::s_infoExtension;
 
   bool data_existed = (myOss.Stat(m_filename.c_str(), &data_stat) == XrdOssOK);
   bool info_existed = (myOss.Stat(ifn.c_str(),        &info_stat) == XrdOssOK);
 
   // Create the data file itself.
   char size_str[32]; sprintf(size_str, "%lld", m_file_size);
   myEnv.Put("oss.asize",  size_str);
   myEnv.Put("oss.cgroup", conf.m_data_space.c_str());
 
   int res;
 
   if ((res = myOss.Create(myUser, m_filename.c_str(), 0600, myEnv, XRDOSS_mkpath)) != XrdOssOK)
   {
      TRACEF(Error, tpfx << "Create failed " << ERRNO_AND_ERRSTR(-res));
      errno = -res;
      return false;
   }
 
   m_data_file = myOss.newFile(myUser);
   if ((res = m_data_file->Open(m_filename.c_str(), O_RDWR, 0600, myEnv)) != XrdOssOK)
   {
      TRACEF(Error, tpfx << "Open failed " << ERRNO_AND_ERRSTR(-res));
      errno = -res;
      delete m_data_file; m_data_file = 0;
      return false;
   }
 
   myEnv.Put("oss.asize", "64k"); // Advisory, block-map and access list lengths vary.
   myEnv.Put("oss.cgroup", conf.m_meta_space.c_str());
   if ((res = myOss.Create(myUser, ifn.c_str(), 0600, myEnv, XRDOSS_mkpath)) != XrdOssOK)
   {
      TRACE(Error, tpfx << "Create failed for info file " << ifn << ERRNO_AND_ERRSTR(-res));
      errno = -res;
      m_data_file->Close(); delete m_data_file; m_data_file = 0;
      return false;
   }
 
   m_info_file = myOss.newFile(myUser);
   if ((res = m_info_file->Open(ifn.c_str(), O_RDWR, 0600, myEnv)) != XrdOssOK)
   {
      TRACEF(Error, tpfx << "Failed for info file " << ifn  << ERRNO_AND_ERRSTR(-res));
      errno = -res;
      delete m_info_file; m_info_file = 0;
      m_data_file->Close(); delete m_data_file;   m_data_file   = 0;
      return false;
   }
 
   bool initialize_info_file = true;
 
   if (info_existed && m_cfi.Read(m_info_file, ifn.c_str()))
   {
      TRACEF(Debug, tpfx << "Reading existing info file. (data_existed=" << data_existed  <<
             ", data_size_stat=" << (data_existed ? data_stat.st_size : -1ll) <<
             ", data_size_from_last_block=" << m_cfi.GetExpectedDataFileSize() <<
             ", block_size=" << (m_cfi.GetBufferSize() >> 10) << "k)");
 
      // Check if data file exists and is of reasonable size.
      if (data_existed && data_stat.st_size >= m_cfi.GetExpectedDataFileSize())
      {
         initialize_info_file = false;
      } else {
         TRACEF(Warning, tpfx << "Basic sanity checks on data file failed, resetting info file, truncating data file.");
         m_cfi.ResetAllAccessStats();
         m_data_file->Ftruncate(0);
         Cache::ResMon().register_file_purge(m_filename, data_stat.st_blocks);
      }
   }
 
   if ( ! initialize_info_file && m_cfi.GetCkSumState() != conf.get_cs_Chk())
   {
      if (conf.does_cschk_have_missing_bits(m_cfi.GetCkSumState()) &&
          conf.should_uvkeep_purge(time(0) - m_cfi.GetNoCkSumTimeForUVKeep()))
      {
         TRACEF(Info, tpfx << "Cksum state of file insufficient, uvkeep test failed, resetting info file, truncating data file.");
         initialize_info_file = true;
         m_cfi.ResetAllAccessStats();
         m_data_file->Ftruncate(0);
         Cache::ResMon().register_file_purge(m_filename, data_stat.st_blocks);
      } else {
         // TODO: If the file is complete, we don't need to reset net cksums.
         m_cfi.DowngradeCkSumState(conf.get_cs_Chk());
      }
   }
 
   // Check if we have pfc url arguments.
   long long pfc_blocksize = conf.m_bufferSize;
   int       pfc_prefetch  = conf.m_prefetch_max_blocks;
   if (conf.m_cgi_blocksize_allowed || conf.m_cgi_prefetch_allowed)
   {
      parse_pfc_url_args(inputIO, pfc_blocksize, pfc_prefetch);
   }
 
   if (initialize_info_file)
   {
      m_cfi.SetBufferSizeFileSizeAndCreationTime(pfc_blocksize, m_file_size);
      m_cfi.SetCkSumState(conf.get_cs_Chk());
      m_cfi.ResetNoCkSumTime();
      m_cfi.Write(m_info_file, ifn.c_str());
      m_info_file->Fsync();
      cache()->WriteFileSizeXAttr(m_info_file->getFD(), m_file_size);
      TRACEF(Debug, tpfx << "Creating new file info, data size = " <<  m_file_size << " num blocks = "  << m_cfi.GetNBlocks()
                         << " block size = " << pfc_blocksize);
   }
   else
   {
      if (futimens(m_info_file->getFD(), NULL)) {
         TRACEF(Error, tpfx << "failed setting modification time " << ERRNO_AND_ERRSTR(errno));
      }
      if (pfc_blocksize != conf.m_bufferSize) {
         TRACEF(Info, tpfx << "URL CGI pfc.blocksize ignored for an already existing file");
      }
   }
 
   m_cfi.WriteIOStatAttach();
   m_state_cond.Lock();
   m_block_size = m_cfi.GetBufferSize();
   m_num_blocks = m_cfi.GetNBlocks();
   m_prefetch_state = (m_cfi.IsComplete()) ? kComplete : kStopped; // Will engage in AddIO().
   m_prefetch_max_blocks_in_flight = pfc_prefetch;
   if (pfc_prefetch != conf.m_prefetch_max_blocks)
      TRACEF(Debug, tpfx << "pfc.prefetch set to " << pfc_prefetch << " via CGI parameter");
 
   m_data_file->Fstat(&data_stat);
   m_st_blocks = data_stat.st_blocks;
 
   m_resmon_token = Cache::ResMon().register_file_open(m_filename, time(0), data_existed);
   constexpr long long MB = 1024 * 1024;
   m_resmon_report_threshold = std::min(std::max(10 * MB, m_file_size / 20), 500 * MB);
   // m_resmon_report_threshold_scaler; // something like 10% of original threshold, to adjust
   // actual threshold based on return values from register_file_update_stats().
 
   m_state_cond.UnLock();
 
   return true;
}
 
void File::parse_pfc_url_args(XrdOucCacheIO* inputIO, long long &pfc_blocksize, int &pfc_prefetch) const
{
   const Configuration &conf = Cache::TheOne().RefConfiguration();
 
   XrdCl::URL url(inputIO->Path());
   auto const & urlp = url.GetParams();
 
   auto extract = [&](const std::string &key, std::string &value) -> bool {
      auto it = urlp.find(key);
      if (it != urlp.end()) {
         value = it->second;
         return true;
      } else {
         value.clear();
         return false;
      }
   };
 
   std::string val;
   if (conf.m_cgi_blocksize_allowed && extract("pfc.blocksize", val))
   {
      const char *tpfx = "File::Open::urlcgi pfc.blocksize ";
      long long bsize;
      if (Cache::TheOne().blocksize_str2value(tpfx, val.c_str(), bsize,
                                              conf.m_cgi_min_bufferSize, conf.m_cgi_max_bufferSize))
      {
         pfc_blocksize = bsize;
      } else {
         TRACEF(Error, tpfx << "Error processing the parameter.");
      }
   }
   if (conf.m_cgi_prefetch_allowed && extract("pfc.prefetch", val))
   {
      const char *tpfx = "File::Open::urlcgi pfc.prefetch ";
      int pref;
      if (Cache::TheOne().prefetch_str2value(tpfx, val.c_str(), pref,
                                             conf.m_cgi_min_prefetch_max_blocks, conf.m_cgi_max_prefetch_max_blocks))
      {
         pfc_prefetch = pref;
      } else {
         TRACEF(Error, tpfx << "Error processing the parameter.");
      }
   }
}
 
//------------------------------------------------------------------------------
 
int File::Fstat(struct stat &sbuff)
{
   // Stat on an open file.
   // Corrects size to actual full size of the file.
   // Sets atime to 0 if the file is only partially downloaded, in accordance
   // with pfc.onlyifcached settings.
   // Called from IO::Fstat() and Cache::Stat() when the file is active.
   // Returns 0 on success, -errno on error.
 
   int res;
 
   if ((res = m_data_file->Fstat(&sbuff))) return res;
 
   sbuff.st_size = m_file_size;
 
   bool is_cached = cache()->DecideIfConsideredCached(m_file_size, sbuff.st_blocks * 512ll);
   if ( ! is_cached)
      sbuff.st_atime = 0;
 
   return 0;
}
 
//==============================================================================
// Read and helpers
//==============================================================================
 
bool File::overlap(int blk,            // block to query
                   long long blk_size, //
                   long long req_off,  // offset of user request
                   int req_size,       // size of user request
                   // output:
                   long long &off,     // offset in user buffer
                   long long &blk_off, // offset in block
                   int       &size)    // size to copy
{
   const long long beg     = blk * blk_size;
   const long long end     = beg + blk_size;
   const long long req_end = req_off + req_size;
 
   if (req_off < end && req_end > beg)
   {
      const long long ovlp_beg = std::max(beg, req_off);
      const long long ovlp_end = std::min(end, req_end);
 
      off     = ovlp_beg - req_off;
      blk_off = ovlp_beg - beg;
      size    = (int) (ovlp_end - ovlp_beg);
 
      assert(size <= blk_size);
      return true;
   }
   else
   {
      return false;
   }
}
 
//------------------------------------------------------------------------------
 
Block* File::PrepareBlockRequest(int i, IO *io, void *req_id, bool prefetch)
{
   // Must be called w/ state_cond locked.
   // Checks on size etc should be done before.
   //
   // Reference count is 0 so increase it in calling function if you want to
   // catch the block while still in memory.
 
   const long long off   = i * m_block_size;
   const int  last_block = m_num_blocks - 1;
   const bool cs_net     = cache()->RefConfiguration().is_cschk_net();
 
   int blk_size, req_size;
   if (i == last_block) {
      blk_size = req_size = m_file_size - off;
      if (cs_net && req_size & 0xFFF) req_size = (req_size & ~0xFFF) + 0x1000;
   } else {
      blk_size = req_size = m_block_size;
   }
 
   Block *b   = 0;
   char  *buf = cache()->RequestRAM(req_size);
 
   if (buf)
   {
      b = new (std::nothrow) Block(this, io, req_id, buf, off, blk_size, req_size, prefetch, cs_net);
 
      if (b)
      {
         m_block_map[i] = b;
 
         // Actual Read request is issued in ProcessBlockRequests().
 
         if (m_prefetch_state == kOn && (int) m_block_map.size() >= m_prefetch_max_blocks_in_flight)
         {
            m_prefetch_state = kHold;
            cache()->DeRegisterPrefetchFile(this);
         }
      }
      else
      {
         TRACEF(Dump, "PrepareBlockRequest() " <<  i << " prefetch " <<  prefetch << ", allocation failed.");
      }
   }
 
   return b;
}
 
void File::ProcessBlockRequest(Block *b)
{
   // This *must not* be called with block_map locked.
 
   BlockResponseHandler* brh = new BlockResponseHandler(b);
 
   if (XRD_TRACE What >= TRACE_Dump) {
      char buf[256];
      snprintf(buf, 256, "idx=%lld, block=%p, prefetch=%d, off=%lld, req_size=%d, buff=%p, resp_handler=%p ",
         b->get_offset()/m_block_size, (void*)b, b->m_prefetch, b->get_offset(), b->get_req_size(), (void*)b->get_buff(), (void*)brh);
      TRACEF(Dump, "ProcessBlockRequest() " << buf);
   }
 
   if (b->req_cksum_net())
   {
      b->get_io()->GetInput()->pgRead(*brh, b->get_buff(), b->get_offset(), b->get_req_size(),
                                      b->ref_cksum_vec(), 0, b->ptr_n_cksum_errors());
   } else {
      b->get_io()->GetInput()->  Read(*brh, b->get_buff(), b->get_offset(), b->get_size());
   }
}
 
void File::ProcessBlockRequests(BlockList_t& blks)
{
   // This *must not* be called with block_map locked.
 
   for (BlockList_i bi = blks.begin(); bi != blks.end(); ++bi)
   {
      ProcessBlockRequest(*bi);
   }
}
 
//------------------------------------------------------------------------------
 
void File::RequestBlocksDirect(IO *io, ReadRequest *read_req, std::vector<XrdOucIOVec>& ioVec, int expected_size)
{
   int n_chunks    = ioVec.size();
   int n_vec_reads = (n_chunks - 1) / XrdProto::maxRvecsz + 1;
 
   TRACEF(DumpXL, "RequestBlocksDirect() issuing ReadV for n_chunks = " << n_chunks <<
          ", total_size = " << expected_size << ", n_vec_reads = " << n_vec_reads);
 
   DirectResponseHandler *handler = new DirectResponseHandler(this, read_req, n_vec_reads);
 
   int pos = 0;
   while (n_chunks > XrdProto::maxRvecsz) {
      io->GetInput()->ReadV( *handler, ioVec.data() + pos, XrdProto::maxRvecsz);
      pos      += XrdProto::maxRvecsz;
      n_chunks -= XrdProto::maxRvecsz;
   }
   io->GetInput()->ReadV( *handler, ioVec.data() + pos, n_chunks);
}
 
//------------------------------------------------------------------------------
 
int File::ReadBlocksFromDisk(std::vector<XrdOucIOVec>& ioVec, int expected_size)
{
   TRACEF(DumpXL, "ReadBlocksFromDisk() issuing ReadV for n_chunks = " << (int) ioVec.size() << ", total_size = " << expected_size);
 
   long long rs = m_data_file->ReadV(ioVec.data(), (int) ioVec.size());
 
   if (rs < 0)
   {
      TRACEF(Error, "ReadBlocksFromDisk neg retval = " <<  rs);
      return rs;
   }
 
   if (rs != expected_size)
   {
      TRACEF(Error, "ReadBlocksFromDisk incomplete size = " << rs);
      return -EIO;
   }
 
   return (int) rs;
}
 
//------------------------------------------------------------------------------
 
int File::Read(IO *io, char* iUserBuff, long long iUserOff, int iUserSize, ReadReqRH *rh)
{
   // rrc_func is ONLY called from async processing.
   // If this function returns anything other than -EWOULDBLOCK, rrc_func needs to be called by the caller.
   // This streamlines implementation of synchronous IO::Read().
 
   TRACEF(Dump, "Read() sid: " << Xrd::hex1 << rh->m_seq_id << " size: " << iUserSize);
 
   m_state_cond.Lock();
 
   if (m_in_shutdown || io->m_in_detach)
   {
      m_state_cond.UnLock();
      return m_in_shutdown ? -ENOENT : -EBADF;
   }
 
   // Shortcut -- file is fully downloaded.
 
   if (m_cfi.IsComplete())
   {
      m_state_cond.UnLock();
      int ret = m_data_file->Read(iUserBuff, iUserOff, iUserSize);
      if (ret > 0) {
         XrdSysCondVarHelper _lck(m_state_cond);
         m_delta_stats.AddBytesHit(ret);
         check_delta_stats();
      }
      return ret;
   }
 
   XrdOucIOVec readV( { iUserOff, iUserSize, 0, iUserBuff } );
 
   return ReadOpusCoalescere(io, &readV, 1, rh, "Read() ");
}
 
//------------------------------------------------------------------------------
 
int File::ReadV(IO *io, const XrdOucIOVec *readV, int readVnum, ReadReqRH *rh)
{
   TRACEF(Dump, "ReadV() for " << readVnum << " chunks.");
 
   m_state_cond.Lock();
 
   if (m_in_shutdown || io->m_in_detach)
   {
      m_state_cond.UnLock();
      return m_in_shutdown ? -ENOENT : -EBADF;
   }
 
   // Shortcut -- file is fully downloaded.
 
   if (m_cfi.IsComplete())
   {
      m_state_cond.UnLock();
      int ret = m_data_file->ReadV(const_cast<XrdOucIOVec*>(readV), readVnum);
      if (ret > 0) {
         XrdSysCondVarHelper _lck(m_state_cond);
         m_delta_stats.AddBytesHit(ret);
         check_delta_stats();
      }
      return ret;
   }
 
   return ReadOpusCoalescere(io, readV, readVnum, rh, "ReadV() ");
}
 
//------------------------------------------------------------------------------
 
int File::ReadOpusCoalescere(IO *io, const XrdOucIOVec *readV, int readVnum,
                             ReadReqRH *rh, const char *tpfx)
{
   // Non-trivial processing for Read and ReadV.
   // Entered under lock.
   //
   // loop over reqired blocks:
   //   - if on disk, ok;
   //   - if in ram or incoming, inc ref-count
   //   - otherwise request and inc ref count (unless RAM full => request direct)
   // unlock
 
   int prefetch_cnt = 0;
 
   ReadRequest *read_req = nullptr;
   BlockList_t  blks_to_request;     // blocks we are issuing a new remote request for
 
   std::unordered_map<Block*, std::vector<ChunkRequest>> blks_ready;
 
   std::vector<XrdOucIOVec> iovec_disk;
   std::vector<XrdOucIOVec> iovec_direct;
   int                      iovec_disk_total = 0;
   int                      iovec_direct_total = 0;
 
   for (int iov_idx = 0; iov_idx < readVnum; ++iov_idx)
   {
      const XrdOucIOVec &iov = readV[iov_idx];
      long long   iUserOff  = iov.offset;
      int         iUserSize = iov.size;
      char       *iUserBuff = iov.data;
 
      const int idx_first = iUserOff / m_block_size;
      const int idx_last  = (iUserOff + iUserSize - 1) / m_block_size;
 
      TRACEF(DumpXL, tpfx << "sid: " << Xrd::hex1 << rh->m_seq_id << " idx_first: " << idx_first << " idx_last: " << idx_last);
 
      enum LastBlock_e { LB_other, LB_disk, LB_direct };
 
      LastBlock_e lbe = LB_other;
 
      for (int block_idx = idx_first; block_idx <= idx_last; ++block_idx)
      {
         TRACEF(DumpXL, tpfx << "sid: " << Xrd::hex1 << rh->m_seq_id << " idx: " << block_idx);
         BlockMap_i bi = m_block_map.find(block_idx);
 
         // overlap and read
         long long off;     // offset in user buffer
         long long blk_off; // offset in block
         int       size;    // size to copy
 
         overlap(block_idx, m_block_size, iUserOff, iUserSize, off, blk_off, size);
 
         // In RAM or incoming?
         if (bi != m_block_map.end())
         {
            inc_ref_count(bi->second);
            TRACEF(Dump, tpfx << (void*) iUserBuff << " inc_ref_count for existing block " << bi->second << " idx = " <<  block_idx);
 
            if (bi->second->is_finished())
            {
               // note, blocks with error should not be here !!!
               // they should be either removed or reissued in ProcessBlockResponse()
               assert(bi->second->is_ok());
 
               blks_ready[bi->second].emplace_back( ChunkRequest(nullptr, iUserBuff + off, blk_off, size) );
 
               if (bi->second->m_prefetch)
                  ++prefetch_cnt;
            }
            else
            {
               if ( ! read_req)
                  read_req = new ReadRequest(io, rh);
 
               // We have a lock on state_cond --> as we register the request before releasing the lock,
               // we are sure to get a call-in via the ChunkRequest handling when this block arrives.
 
               bi->second->m_chunk_reqs.emplace_back( ChunkRequest(read_req, iUserBuff + off, blk_off, size) );
               ++read_req->m_n_chunk_reqs;
            }
 
            lbe = LB_other;
         }
         // On disk?
         else if (m_cfi.TestBitWritten(offsetIdx(block_idx)))
         {
            TRACEF(DumpXL, tpfx << "read from disk " <<  (void*)iUserBuff << " idx = " << block_idx);
 
            if (lbe == LB_disk)
               iovec_disk.back().size += size;
            else
               iovec_disk.push_back( { block_idx * m_block_size + blk_off, size, 0, iUserBuff + off } );
            iovec_disk_total += size;
 
            if (m_cfi.TestBitPrefetch(offsetIdx(block_idx)))
               ++prefetch_cnt;
 
            lbe = LB_disk;
         }
         // Neither ... then we have to go get it ...
         else
         {
            if ( ! read_req)
               read_req = new ReadRequest(io, rh);
 
            // Is there room for one more RAM Block?
            Block *b = PrepareBlockRequest(block_idx, io, read_req, false);
            if (b)
            {
               TRACEF(Dump, tpfx << "inc_ref_count new " <<  (void*)iUserBuff << " idx = " << block_idx);
               inc_ref_count(b);
               blks_to_request.push_back(b);
 
               b->m_chunk_reqs.emplace_back(ChunkRequest(read_req, iUserBuff + off, blk_off, size));
               ++read_req->m_n_chunk_reqs;
 
               lbe = LB_other;
            }
            else // Nope ... read this directly without caching.
            {
               TRACEF(DumpXL, tpfx << "direct block " << block_idx << ", blk_off " << blk_off << ", size " << size);
 
               iovec_direct_total += size;
               read_req->m_direct_done = false;
 
               // Make sure we do not issue a ReadV with chunk size above XrdProto::maxRVdsz.
               // Number of actual ReadVs issued so as to not exceed the XrdProto::maxRvecsz limit
               // is determined in the RequestBlocksDirect().
               if (lbe == LB_direct && iovec_direct.back().size + size <= XrdProto::maxRVdsz) {
                  iovec_direct.back().size += size;
               } else {
                  long long  in_offset = block_idx * m_block_size + blk_off;
                  char      *out_pos   = iUserBuff + off;
                  while (size > XrdProto::maxRVdsz) {
                     iovec_direct.push_back( { in_offset, XrdProto::maxRVdsz, 0, out_pos } );
                     in_offset += XrdProto::maxRVdsz;
                     out_pos   += XrdProto::maxRVdsz;
                     size      -= XrdProto::maxRVdsz;
                  }
                  iovec_direct.push_back( { in_offset, size, 0, out_pos } );
               }
 
               lbe = LB_direct;
            }
         }
      } // end for over blocks in an IOVec
   } // end for over readV IOVec
 
   inc_prefetch_hit_cnt(prefetch_cnt);
 
   m_state_cond.UnLock();
 
   // First, send out remote requests for new blocks.
   if ( ! blks_to_request.empty())
   {
      ProcessBlockRequests(blks_to_request);
      blks_to_request.clear();
   }
 
   // Second, send out remote direct read requests.
   if ( ! iovec_direct.empty())
   {
      RequestBlocksDirect(io, read_req, iovec_direct, iovec_direct_total);
 
      TRACEF(Dump, tpfx << "direct read requests sent out, n_chunks = " << (int) iovec_direct.size() << ", total_size = " << iovec_direct_total);
   }
 
   // Begin synchronous part where we process data that is already in RAM or on disk.
 
   long long bytes_read = 0;
   int       error_cond = 0; // to be set to -errno
 
   // Third, process blocks that are available in RAM.
   if ( ! blks_ready.empty())
   {
      for (auto &bvi : blks_ready)
      {
         for (auto &cr : bvi.second)
         {
            TRACEF(DumpXL, tpfx << "ub=" << (void*)cr.m_buf << " from pre-finished block " << bvi.first->m_offset/m_block_size << " size " << cr.m_size);
            memcpy(cr.m_buf, bvi.first->m_buff + cr.m_off, cr.m_size);
            bytes_read += cr.m_size;
         }
      }
   }
 
   // Fourth, read blocks from disk.
   if ( ! iovec_disk.empty())
   {
      int rc = ReadBlocksFromDisk(iovec_disk, iovec_disk_total);
      TRACEF(DumpXL, tpfx << "from disk finished size = " << rc);
      if (rc >= 0)
      {
         bytes_read += rc;
      }
      else
      {
         error_cond = rc;
         TRACEF(Error, tpfx << "failed read from disk");
      }
   }
 
   // End synchronous part -- update with sync stats and determine actual state of this read.
   // Note: remote reads might have already finished during disk-read!
 
   m_state_cond.Lock();
 
   for (auto &bvi : blks_ready)
      dec_ref_count(bvi.first, (int) bvi.second.size());
 
   if (read_req)
   {
      read_req->m_bytes_read += bytes_read;
      if (error_cond)
         read_req->update_error_cond(error_cond);
      read_req->m_stats.m_BytesHit += bytes_read;
      read_req->m_sync_done = true;
 
      if (read_req->is_complete())
      {
         // Almost like FinalizeReadRequest(read_req) -- but no callout!
         m_delta_stats.AddReadStats(read_req->m_stats);
         check_delta_stats();
         m_state_cond.UnLock();
 
         int ret = read_req->return_value();
         delete read_req;
         return ret;
      }
      else
      {
         m_state_cond.UnLock();
         return -EWOULDBLOCK;
      }
   }
   else
   {
      m_delta_stats.m_BytesHit += bytes_read;
      check_delta_stats();
      m_state_cond.UnLock();
 
      // !!! No callout.
 
      return error_cond ? error_cond : bytes_read;
   }
}
 
 
//==============================================================================
// WriteBlock and Sync
//==============================================================================
 
void File::WriteBlockToDisk(Block* b)
{
   // write block buffer into disk file
   long long   offset = b->m_offset - m_offset;
   long long   size   = b->get_size();
   ssize_t     retval;
 
   if (m_cfi.IsCkSumCache())
      if (b->has_cksums())
         retval = m_data_file->pgWrite(b->get_buff(), offset, size, b->ref_cksum_vec().data(), 0);
      else
         retval = m_data_file->pgWrite(b->get_buff(), offset, size, 0, 0);
   else
      retval = m_data_file->Write(b->get_buff(), offset, size);
 
   if (retval < size)
   {
      if (retval < 0) {
         TRACEF(Error, "WriteToDisk() write error " << retval);
      } else {
         TRACEF(Error, "WriteToDisk() incomplete block write ret=" << retval << " (should be " << size << ")");
      }
 
      XrdSysCondVarHelper _lck(m_state_cond);
 
      dec_ref_count(b);
 
      return;
   }
 
   const int blk_idx =  (b->m_offset - m_offset) / m_block_size;
 
   // Set written bit.
   TRACEF(Dump, "WriteToDisk() success set bit for block " <<  b->m_offset << " size=" <<  size);
 
   bool schedule_sync = false;
   {
      XrdSysCondVarHelper _lck(m_state_cond);
 
      m_cfi.SetBitWritten(blk_idx);
 
      if (b->m_prefetch)
      {
         m_cfi.SetBitPrefetch(blk_idx);
      }
      if (b->req_cksum_net() && ! b->has_cksums() && m_cfi.IsCkSumNet())
      {
         m_cfi.ResetCkSumNet();
      }
 
      // Set synced bit or stash block index if in actual sync.
      // Synced state is only written out to cinfo file when data file is synced.
      if (m_in_sync)
      {
         m_writes_during_sync.push_back(blk_idx);
      }
      else
      {
         m_cfi.SetBitSynced(blk_idx);
         ++m_non_flushed_cnt;
         if ((m_cfi.IsComplete() || m_non_flushed_cnt >= Cache::GetInstance().RefConfiguration().m_flushCnt) &&
             ! m_in_shutdown)
         {
            schedule_sync     = true;
            m_in_sync         = true;
            m_non_flushed_cnt = 0;
         }
      }
      // As soon as the reference count is decreased on the block, the
      // file object may be deleted.  Thus, to avoid holding both locks at a time,
      // we defer the ref count decrease until later if a sync is needed
      if (!schedule_sync) {
         dec_ref_count(b);
      }
   }
 
   if (schedule_sync)
   {
      cache()->ScheduleFileSync(this);
      XrdSysCondVarHelper _lck(m_state_cond);
      dec_ref_count(b);
   }
}
 
//------------------------------------------------------------------------------
 
void File::Sync()
{
   TRACEF(Dump, "Sync()");
 
   int ret     = m_data_file->Fsync();
   bool errorp = false;
   if (ret == XrdOssOK)
   {
      Stats loc_stats;
      {
         XrdSysCondVarHelper _lck(&m_state_cond);
         report_and_merge_delta_stats();
         loc_stats = m_stats;
      }
      m_cfi.WriteIOStat(loc_stats);
      m_cfi.Write(m_info_file, m_filename.c_str());
      int cret = m_info_file->Fsync();
      if (cret != XrdOssOK)
      {
         TRACEF(Error, "Sync cinfo file sync error " << cret);
         errorp = true;
      }
   }
   else
   {
      TRACEF(Error, "Sync data file sync error " << ret << ", cinfo file has not been updated");
      errorp = true;
   }
 
   if (errorp)
   {
      TRACEF(Error, "Sync failed, unlinking local files and initiating shutdown of File object");
 
      // Unlink will also call this->initiate_emergency_shutdown()
      Cache::GetInstance().UnlinkFile(m_filename, false);
 
      XrdSysCondVarHelper _lck(&m_state_cond);
 
      m_writes_during_sync.clear();
      m_in_sync = false;
 
      return;
   }
 
   int  written_while_in_sync;
   bool resync = false;
   {
      XrdSysCondVarHelper _lck(&m_state_cond);
      for (std::vector<int>::iterator i = m_writes_during_sync.begin(); i != m_writes_during_sync.end(); ++i)
      {
         m_cfi.SetBitSynced(*i);
      }
      written_while_in_sync = m_non_flushed_cnt = (int) m_writes_during_sync.size();
      m_writes_during_sync.clear();
 
      // If there were writes during sync and the file is now complete,
      // let us call Sync again without resetting the m_in_sync flag.
      if (written_while_in_sync > 0 && m_cfi.IsComplete() && ! m_in_shutdown)
         resync = true;
      else
         m_in_sync = false;
   }
   TRACEF(Dump, "Sync "<< written_while_in_sync  << " blocks written during sync." << (resync ? " File is now complete - resyncing." : ""));
 
   if (resync)
      Sync();
}
 
 
//==============================================================================
// Block processing
//==============================================================================
 
void File::free_block(Block* b)
{
   // Method always called under lock.
   int i = b->m_offset / m_block_size;
   TRACEF(Dump, "free_block block " << b << "  idx =  " <<  i);
   size_t ret = m_block_map.erase(i);
   if (ret != 1)
   {
      // assert might be a better option than a warning
      TRACEF(Error, "free_block did not erase " <<  i  << " from map");
   }
   else
   {
      cache()->ReleaseRAM(b->m_buff, b->m_req_size);
      delete b;
   }
 
   if (m_prefetch_state == kHold && (int) m_block_map.size() < m_prefetch_max_blocks_in_flight)
   {
      m_prefetch_state = kOn;
      cache()->RegisterPrefetchFile(this);
   }
}
 
//------------------------------------------------------------------------------
 
bool File::select_current_io_or_disable_prefetching(bool skip_current)
{
   // Method always called under lock. It also expects prefetch to be active.
 
   int  io_size = (int) m_io_set.size();
   bool io_ok   = false;
 
   if (io_size == 1)
   {
      io_ok = (*m_io_set.begin())->m_allow_prefetching;
      if (io_ok)
      {
         m_current_io = m_io_set.begin();
      }
   }
   else if (io_size > 1)
   {
      IoSet_i mi = m_current_io;
      if (skip_current && mi != m_io_set.end()) ++mi;
 
      for (int i = 0; i < io_size; ++i)
      {
         if (mi == m_io_set.end()) mi = m_io_set.begin();
 
         if ((*mi)->m_allow_prefetching)
         {
            m_current_io = mi;
            io_ok = true;
            break;
         }
         ++mi;
      }
   }
 
   if ( ! io_ok)
   {
      m_current_io    = m_io_set.end();
      m_prefetch_state = kStopped;
      cache()->DeRegisterPrefetchFile(this);
   }
 
   return io_ok;
}
 
//------------------------------------------------------------------------------
 
void File::ProcessDirectReadFinished(ReadRequest *rreq, int bytes_read, int error_cond)
{
   // Called from DirectResponseHandler.
   // NOT under lock.
 
   if (error_cond)
      TRACEF(Error, "Read(), direct read finished with error " << -error_cond << " " << XrdSysE2T(-error_cond));
 
   m_state_cond.Lock();
 
   if (error_cond)
      rreq->update_error_cond(error_cond);
   else {
      rreq->m_stats.m_BytesBypassed += bytes_read;
      rreq->m_bytes_read += bytes_read;
   }
 
   rreq->m_direct_done = true;
 
   bool rreq_complete = rreq->is_complete();
 
   m_state_cond.UnLock();
 
   if (rreq_complete)
      FinalizeReadRequest(rreq);
}
 
void File::ProcessBlockError(Block *b, ReadRequest *rreq)
{
   // Called from ProcessBlockResponse().
   // YES under lock -- we have to protect m_block_map for recovery through multiple IOs.
   // Does not manage m_read_req.
   // Will not complete the request.
 
   TRACEF(Debug, "ProcessBlockError() io " << b->m_io << ", block "<< b->m_offset/m_block_size <<
                 " finished with error " << -b->get_error() << " " << XrdSysE2T(-b->get_error()));
 
   rreq->update_error_cond(b->get_error());
   --rreq->m_n_chunk_reqs;
 
   dec_ref_count(b);
}
 
void File::ProcessBlockSuccess(Block *b, ChunkRequest &creq)
{
   // Called from ProcessBlockResponse().
   // NOT under lock as it does memcopy ofor exisf block data.
   // Acquires lock for block, m_read_req and rreq state update.
 
   ReadRequest *rreq = creq.m_read_req;
 
   TRACEF(Dump, "ProcessBlockSuccess() ub=" << (void*)creq.m_buf  << " from finished block " << b->m_offset/m_block_size << " size " << creq.m_size);
   memcpy(creq.m_buf, b->m_buff + creq.m_off, creq.m_size);
 
   m_state_cond.Lock();
 
   rreq->m_bytes_read += creq.m_size;
 
   if (b->get_req_id() == (void*) rreq)
      rreq->m_stats.m_BytesMissed += creq.m_size;
   else
      rreq->m_stats.m_BytesHit    += creq.m_size;
 
   --rreq->m_n_chunk_reqs;
 
   if (b->m_prefetch)
      inc_prefetch_hit_cnt(1);
 
   dec_ref_count(b);
 
   bool rreq_complete = rreq->is_complete();
 
   m_state_cond.UnLock();
 
   if (rreq_complete)
      FinalizeReadRequest(rreq);
}
 
void File::FinalizeReadRequest(ReadRequest *rreq)
{
   // called from ProcessBlockResponse()
   // NOT under lock -- does callout
   {
      XrdSysCondVarHelper _lck(m_state_cond);
      m_delta_stats.AddReadStats(rreq->m_stats);
      check_delta_stats();
   }
 
   rreq->m_rh->Done(rreq->return_value());
   delete rreq;
}
 
void File::ProcessBlockResponse(Block *b, int res)
{
   static const char* tpfx = "ProcessBlockResponse ";
 
   TRACEF(Dump, tpfx << "block=" << b << ", idx=" << b->m_offset/m_block_size << ", off=" << b->m_offset << ", res=" << res);
 
   if (res >= 0 && res != b->get_size())
   {
      // Incorrect number of bytes received, apparently size of the file on the remote
      // is different than what the cache expects it to be.
      TRACEF(Error, tpfx << "Wrong number of bytes received, assuming remote/local file size mismatch, unlinking local files and initiating shutdown of File object");
      Cache::GetInstance().UnlinkFile(m_filename, false);
   }
 
   m_state_cond.Lock();
 
   // Deregister block from IO's prefetch count, if needed.
   if (b->m_prefetch)
   {
      IO     *io = b->get_io();
      IoSet_i mi = m_io_set.find(io);
      if (mi != m_io_set.end())
      {
         --io->m_active_prefetches;
 
         // If failed and IO is still prefetching -- disable prefetching on this IO.
         if (res < 0 && io->m_allow_prefetching)
         {
            TRACEF(Debug, tpfx << "after failed prefetch on io " << io << " disabling prefetching on this io.");
            io->m_allow_prefetching = false;
 
            // Check if any IO is still available for prfetching. If not, stop it.
            if (m_prefetch_state == kOn || m_prefetch_state == kHold)
            {
               if ( ! select_current_io_or_disable_prefetching(false) )
               {
                  TRACEF(Debug, tpfx << "stopping prefetching after io " <<  b->get_io() << " marked as bad.");
               }
            }
         }
 
         // If failed with no subscribers -- delete the block and exit.
         if (b->m_refcnt == 0 && (res < 0 || m_in_shutdown))
         {
            free_block(b);
            m_state_cond.UnLock();
            return;
         }
         m_prefetch_bytes += b->get_size();
      }
      else
      {
         TRACEF(Error, tpfx << "io " << b->get_io() << " not found in IoSet.");
      }
   }
 
   if (res == b->get_size())
   {
      b->set_downloaded();
      TRACEF(Dump, tpfx << "inc_ref_count idx=" <<  b->m_offset/m_block_size);
      if ( ! m_in_shutdown)
      {
         // Increase ref-count for the writer.
         inc_ref_count(b);
         m_delta_stats.AddWriteStats(b->get_size(), b->get_n_cksum_errors());
         // No check for writes, report-and-merge forced during Sync().
         cache()->AddWriteTask(b, true);
      }
 
      // Swap chunk-reqs vector out of Block, it will be processed outside of lock.
      vChunkRequest_t  creqs_to_notify;
      creqs_to_notify.swap( b->m_chunk_reqs );
 
      m_state_cond.UnLock();
 
      for (auto &creq : creqs_to_notify)
      {
         ProcessBlockSuccess(b, creq);
      }
   }
   else
   {
      if (res < 0) {
         bool new_error = b->get_io()->register_block_error(res);
         int tlvl = new_error ? TRACE_Error : TRACE_Debug;
         TRACEF_INT(tlvl, tpfx << "block " << b << ", idx=" << b->m_offset/m_block_size << ", off=" << b->m_offset
                    << ", io=" <<  b->get_io() << ", error=" << res);
      } else {
         bool first_p = b->get_io()->register_incomplete_read();
         int tlvl = first_p ? TRACE_Error : TRACE_Debug;
         TRACEF_INT(tlvl, tpfx << "block " << b << ", idx=" << b->m_offset/m_block_size << ", off=" << b->m_offset
                    << ", io=" <<  b->get_io() << " incomplete, got " << res << " expected " << b->get_size());
#if defined(__APPLE__) || defined(__GNU__) || (defined(__FreeBSD_kernel__) && defined(__GLIBC__)) || defined(__FreeBSD__)
         res = -EIO;
#else
         res = -EREMOTEIO;
#endif
      }
      b->set_error(res);
 
      // Loop over Block's chunk-reqs vector, error out ones with the same IO.
      // Collect others with a different IO, the first of them will be used to reissue the request.
      // This is then done outside of lock.
      std::list<ReadRequest*> rreqs_to_complete;
      vChunkRequest_t         creqs_to_keep;
 
      for(ChunkRequest &creq : b->m_chunk_reqs)
      {
         ReadRequest *rreq = creq.m_read_req;
 
         if (rreq->m_io == b->get_io())
         {
            ProcessBlockError(b, rreq);
            if (rreq->is_complete())
            {
               rreqs_to_complete.push_back(rreq);
            }
         }
         else
         {
            creqs_to_keep.push_back(creq);
         }
      }
 
      bool reissue = false;
      if ( ! creqs_to_keep.empty())
      {
         ReadRequest *rreq = creqs_to_keep.front().m_read_req;
 
         TRACEF(Debug, "ProcessBlockResponse() requested block " << (void*)b << " failed with another io " <<
               b->get_io() << " - reissuing request with my io " << rreq->m_io);
 
         b->reset_error_and_set_io(rreq->m_io, rreq);
         b->m_chunk_reqs.swap( creqs_to_keep );
         reissue = true;
      }
 
      m_state_cond.UnLock();
 
      for (auto rreq : rreqs_to_complete)
         FinalizeReadRequest(rreq);
 
      if (reissue)
         ProcessBlockRequest(b);
   }
}
 
//------------------------------------------------------------------------------
 
const char* File::lPath() const
{
   return m_filename.c_str();
}
 
//------------------------------------------------------------------------------
 
int File::offsetIdx(int iIdx) const
{
   return iIdx - m_offset/m_block_size;
}
 
 
//------------------------------------------------------------------------------
 
void File::Prefetch()
{
   // Check that block is not on disk and not in RAM.
   // TODO: Could prefetch several blocks at once!
   //       blks_max could be an argument
 
   BlockList_t blks;
 
   TRACEF(DumpXL, "Prefetch() entering.");
   {
      XrdSysCondVarHelper _lck(m_state_cond);
 
      if (m_prefetch_state != kOn)
      {
         return;
      }
 
      if ( ! select_current_io_or_disable_prefetching(true) )
      {
         TRACEF(Error, "Prefetch no available IO object found, prefetching stopped. This should not happen, i.e., prefetching should be stopped before.");
         return;
      }
 
      // Select block(s) to fetch.
      for (int f = 0; f < m_num_blocks; ++f)
      {
         if ( ! m_cfi.TestBitWritten(f))
         {
            int f_act = f + m_offset / m_block_size;
 
            BlockMap_i bi = m_block_map.find(f_act);
            if (bi == m_block_map.end())
            {
               Block *b = PrepareBlockRequest(f_act, *m_current_io, nullptr, true);
               if (b)
               {
                  TRACEF(Dump, "Prefetch take block " << f_act);
                  blks.push_back(b);
                  // Note: block ref_cnt not increased, it will be when placed into write queue.
 
                  inc_prefetch_read_cnt(1);
               }
               else
               {
                  // This shouldn't happen as prefetching stops when RAM is 70% full.
                  TRACEF(Warning, "Prefetch allocation failed for block " << f_act);
               }
               break;
            }
         }
      }
 
      if (blks.empty())
      {
         TRACEF(Debug, "Prefetch file is complete, stopping prefetch.");
         m_prefetch_state = kComplete;
         cache()->DeRegisterPrefetchFile(this);
      }
      else
      {
         (*m_current_io)->m_active_prefetches += (int) blks.size();
      }
   }
 
   if ( ! blks.empty())
   {
      ProcessBlockRequests(blks);
   }
}
 
 
//------------------------------------------------------------------------------
 
float File::GetPrefetchScore() const
{
   return m_prefetch_score;
}
 
XrdSysError* File::GetLog() const
{
   return Cache::TheOne().GetLog();
}
 
XrdSysTrace* File::GetTrace() const
{
   return Cache::TheOne().GetTrace();
}
 
void File::insert_remote_location(const std::string &loc)
{
   if ( ! loc.empty())
   {
      size_t p = loc.find_first_of('@');
      m_remote_locations.insert(&loc[p != std::string::npos ? p + 1 : 0]);
   }
}
 
std::string File::GetRemoteLocations() const
{
   std::string s;
   if ( ! m_remote_locations.empty())
   {
      size_t      sl = 0;
      int         nl = 0;
      for (std::set<std::string>::iterator i = m_remote_locations.begin(); i != m_remote_locations.end(); ++i, ++nl)
      {
         sl += i->size();
      }
      s.reserve(2 + sl + 2*nl + nl - 1 + 1);
      s = '[';
      int j = 1;
      for (std::set<std::string>::iterator i = m_remote_locations.begin(); i != m_remote_locations.end(); ++i, ++j)
      {
         s += '"'; s += *i; s += '"';
         if (j < nl) s += ',';
      }
      s += ']';
   }
   else
   {
      s = "[]";
   }
   return s;
}
 
//==============================================================================
//=======================    RESPONSE HANDLERS    ==============================
//==============================================================================
 
void BlockResponseHandler::Done(int res)
{
   m_block->m_file->ProcessBlockResponse(m_block, res);
   delete this;
}
 
//------------------------------------------------------------------------------
 
void DirectResponseHandler::Done(int res)
{
   m_mutex.Lock();
 
   int n_left = --m_to_wait;
 
   if (res < 0) {
      if (m_errno == 0) m_errno = res; // store first reported error
   } else {
      m_bytes_read += res;
   }
 
   m_mutex.UnLock();
 
   if (n_left == 0)
   {
      m_file->ProcessDirectReadFinished(m_read_req, m_bytes_read, m_errno);
      delete this;
   }
}