XrdCl::ClassicCopyJob Class Reference

#include <XrdClClassicCopyJob.hh>

Inheritance diagram for XrdCl::ClassicCopyJob:

Collaboration diagram for XrdCl::ClassicCopyJob:

Public Member Functions
	ClassicCopyJob (uint16_t jobId, PropertyList *jobProperties, PropertyList *jobResults)
const XRootDStatus &	GetResult () const
virtual XRootDStatus	Run (CopyProgressHandler *progress=0)
Public Member Functions inherited from XrdCl::CopyJob
	CopyJob (uint16_t jobId, PropertyList *jobProperties, PropertyList *jobResults)
	Constructor. More...
virtual	~CopyJob ()
	Virtual destructor. More...
PropertyList *	GetProperties ()
	Get the job properties. More...
PropertyList *	GetResults ()
	Get the job results. More...
const URL &	GetSource () const
	Get source. More...
const URL &	GetTarget () const
	Get target. More...
void	Init ()

Additional Inherited Members
Protected Attributes inherited from XrdCl::CopyJob
uint16_t	pJobId
PropertyList *	pProperties
PropertyList *	pResults
URL	pSource
URL	pTarget

Detailed Description

Definition at line 27 of file XrdClClassicCopyJob.hh.

Constructor & Destructor Documentation

ClassicCopyJob()

XrdCl::ClassicCopyJob::ClassicCopyJob	(	uint16_t	jobId,
		PropertyList *	jobProperties,
		PropertyList *	jobResults
	)

Definition at line 2416 of file XrdClClassicCopyJob.cc.

                                                            :
    CopyJob( jobId, jobProperties, jobResults )
  {
    Log *log = DefaultEnv::GetLog();
    log->Debug( UtilityMsg, "Creating a classic copy job, from %s to %s",
                GetSource().GetObfuscatedURL().c_str(), GetTarget().GetObfuscatedURL().c_str() );
  }

References XrdCl::Log::Debug(), XrdCl::DefaultEnv::GetLog(), XrdCl::CopyJob::GetSource(), XrdCl::CopyJob::GetTarget(), and XrdCl::UtilityMsg.

Here is the call graph for this function:

Member Function Documentation

GetResult()

const XRootDStatus& XrdCl::ClassicCopyJob::GetResult ( ) const

inline

Definition at line 48 of file XrdClClassicCopyJob.hh.

      {
        return result;
      }

Run()

XRootDStatus XrdCl::ClassicCopyJob::Run ( CopyProgressHandler *  progress = 0 )

virtual

Run the copy job

Parameters

progress

the handler to be notified about the copy progress

Returns

status of the copy operation

Implements XrdCl::CopyJob.

Definition at line 2429 of file XrdClClassicCopyJob.cc.

  {
    Log *log = DefaultEnv::GetLog();
 
    std::string checkSumMode;
    std::string checkSumType;
    std::string checkSumPreset;
    std::string zipSource;
    uint16_t    parallelChunks;
    uint32_t    chunkSize;
    uint64_t    blockSize;
    bool        posc, force, coerce, makeDir, dynamicSource, zip, xcp, preserveXAttr,
                rmOnBadCksum, continue_, zipappend, doserver;
    int32_t     nbXcpSources;
    long long   xRate;
    long long   xRateThreshold;
    uint16_t    cpTimeout;
    std::vector<std::string> addcksums;
 
    pProperties->Get( "checkSumMode",    checkSumMode );
    pProperties->Get( "checkSumType",    checkSumType );
    pProperties->Get( "checkSumPreset",  checkSumPreset );
    pProperties->Get( "parallelChunks",  parallelChunks );
    pProperties->Get( "chunkSize",       chunkSize );
    pProperties->Get( "posc",            posc );
    pProperties->Get( "force",           force );
    pProperties->Get( "coerce",          coerce );
    pProperties->Get( "makeDir",         makeDir );
    pProperties->Get( "dynamicSource",   dynamicSource );
    pProperties->Get( "zipArchive",      zip );
    pProperties->Get( "xcp",             xcp );
    pProperties->Get( "xcpBlockSize",    blockSize );
    pProperties->Get( "preserveXAttr",   preserveXAttr );
    pProperties->Get( "xrate",           xRate );
    pProperties->Get( "xrateThreshold",  xRateThreshold );
    pProperties->Get( "rmOnBadCksum",    rmOnBadCksum );
    pProperties->Get( "continue",        continue_ );
    pProperties->Get( "cpTimeout",       cpTimeout );
    pProperties->Get( "zipAppend",       zipappend );
    pProperties->Get( "addcksums",       addcksums );
    pProperties->Get( "doServer",        doserver );
 
    if( zip )
      pProperties->Get( "zipSource",     zipSource );
 
    if( xcp )
      pProperties->Get( "nbXcpSources",  nbXcpSources );
 
    if( force && continue_ )
      return SetResult( stError, errInvalidArgs, EINVAL,
                     "Invalid argument combination: continue + force." );
 
    if( zipappend && ( continue_ || force ) )
      return SetResult( stError, errInvalidArgs, EINVAL,
                     "Invalid argument combination: ( continue | force ) + zip-append." );
 
    //--------------------------------------------------------------------------
    // Start the cp t/o timer if necessary
    //--------------------------------------------------------------------------
    std::unique_ptr<timer_sec_t> cptimer;
    if( cpTimeout ) cptimer.reset( new timer_sec_t() );
 
    //--------------------------------------------------------------------------
    // Remove on bad checksum implies that POSC semantics has to be enabled
    //--------------------------------------------------------------------------
    if( rmOnBadCksum ) posc = true;
 
    //--------------------------------------------------------------------------
    // Resolve the 'auto' checksum type.
    //--------------------------------------------------------------------------
    if( checkSumType == "auto" )
    {
      checkSumType = Utils::InferChecksumType( GetSource(), GetTarget(), zip );
      if( checkSumType.empty() )
        return SetResult( stError, errCheckSumError, ENOTSUP, "Could not infer checksum type." );
      else
        log->Info( UtilityMsg, "Using inferred checksum type: %s.", checkSumType.c_str() );
    }
 
    if( cptimer && cptimer->elapsed() > cpTimeout ) // check the CP timeout
      return SetResult( stError, errOperationExpired, 0, "CPTimeout exceeded." );
 
    //--------------------------------------------------------------------------
    // Initialize the source and the destination
    //--------------------------------------------------------------------------
    std::unique_ptr<Source> src;
    if( xcp )
      src.reset( new XRootDSourceXCp( &GetSource(), chunkSize, parallelChunks, nbXcpSources, blockSize ) );
    else if( zip ) // TODO make zip work for xcp
      src.reset( new XRootDSourceZip( zipSource, &GetSource(), chunkSize, parallelChunks,
                                      checkSumType, addcksums , doserver) );
    else if( GetSource().GetProtocol() == "stdio" )
      src.reset( new StdInSource( checkSumType, chunkSize, addcksums ) );
    else
    {
      if( dynamicSource )
        src.reset( new XRootDSourceDynamic( &GetSource(), chunkSize, checkSumType, addcksums ) );
      else
        src.reset( new XRootDSource( &GetSource(), chunkSize, parallelChunks, checkSumType, addcksums, doserver ) );
    }
 
    XRootDStatus st = src->Initialize();
    if( !st.IsOK() ) return SourceError( st );
    uint64_t size = src->GetSize() >= 0 ? src->GetSize() : 0;
 
    if( cptimer && cptimer->elapsed() > cpTimeout ) // check the CP timeout
      return SetResult( stError, errOperationExpired, 0, "CPTimeout exceeded." );
 
    std::unique_ptr<Destination> dest;
    URL newDestUrl( GetTarget() );
 
    if( GetTarget().GetProtocol() == "stdio" )
      dest.reset( new StdOutDestination( checkSumType ) );
    else if( zipappend )
    {
      std::string fn = GetSource().GetPath();
      size_t pos = fn.rfind( '/' );
      if( pos != std::string::npos )
        fn = fn.substr( pos + 1 );
      int64_t size = src->GetSize();
      dest.reset( new XRootDZipDestination( newDestUrl, fn, size, parallelChunks, *this ) );
    }
    //--------------------------------------------------------------------------
    // For xrootd destination build the oss.asize hint
    //--------------------------------------------------------------------------
    else
    {
      if( src->GetSize() >= 0 )
      {
        URL::ParamsMap params = newDestUrl.GetParams();
        std::ostringstream o; o << src->GetSize();
        params["oss.asize"] = o.str();
        newDestUrl.SetParams( params );
 //     makeDir = true; // Backward compatibility for xroot destinations!!!
      }
      dest.reset( new XRootDDestination( newDestUrl, parallelChunks, checkSumType, *this ) );
    }
 
    dest->SetForce( force );
    dest->SetPOSC(  posc );
    dest->SetCoerce( coerce );
    dest->SetMakeDir( makeDir );
    dest->SetContinue( continue_ );
    st = dest->Initialize();
    if( !st.IsOK() ) return DestinationError( st );
 
    if( cptimer && cptimer->elapsed() > cpTimeout ) // check the CP timeout
      return SetResult( stError, errOperationExpired, 0, "CPTimeout exceeded." );
 
    //--------------------------------------------------------------------------
    // Copy the chunks
    //--------------------------------------------------------------------------
    if( continue_ )
    {
      size -= dest->GetSize();
      XrdCl::XRootDStatus st = src->StartAt( dest->GetSize() );
      if( !st.IsOK() ) return SetResult( st );
    }
 
    PageInfo  pageInfo;
    uint64_t  total_processed = 0;
    uint64_t  processed = 0;
    auto      start = time_nsec();
    uint16_t  threshold_interval = parallelChunks;
    bool      threshold_draining = false;
    timer_nsec_t threshold_timer;
    while( 1 )
    {
      st = src->GetChunk( pageInfo );
      if( !st.IsOK() )
        return SourceError( st);
 
      if( st.IsOK() && st.code == suDone )
        break;
 
      if( cptimer && cptimer->elapsed() > cpTimeout ) // check the CP timeout
        return SetResult( stError, errOperationExpired, 0, "CPTimeout exceeded." );
 
      if( xRate )
      {
        auto   elapsed     = ( time_nsec() - start ).count();
        double transferred = total_processed + pageInfo.GetLength();
        double expected    = double( xRate ) / to_nsec( 1 ) * elapsed;
        //----------------------------------------------------------------------
        // check if our transfer rate didn't exceeded the limit
        // (we are too fast)
        //----------------------------------------------------------------------
        if( elapsed && // make sure elapsed time is greater than 0
            transferred > expected )
        {
          auto nsec = ( transferred / xRate * to_nsec( 1 ) ) - elapsed;
          sleep_nsec( nsec );
        }
      }
 
      if( xRateThreshold )
      {
        auto   elapsed     = threshold_timer.elapsed();
        double transferred = processed + pageInfo.GetLength();
        double expected    = double( xRateThreshold ) / to_nsec( 1 ) * elapsed;
        //----------------------------------------------------------------------
        // check if our transfer rate dropped below the threshold
        // (we are too slow)
        //----------------------------------------------------------------------
        if( elapsed && // make sure elapsed time is greater than 0
            transferred < expected &&
            threshold_interval == 0 ) // we check every # parallelChunks
        {
          if( !threshold_draining )
          {
            log->Warning( UtilityMsg, "Transfer rate dropped below requested ehreshold,"
                                      " trying different source!" );
            XRootDStatus st = src->TryOtherServer();
            if( !st.IsOK() ) return SetResult( stError, errThresholdExceeded, 0,
                                            "The transfer rate dropped below "
                                            "requested threshold!" );
            threshold_draining = true; // before the next measurement we need to drain
                                       // all the chunks that will come from the old server
          }
          else // now that all the chunks from the old server have
          {    // been received we can start another measurement
            processed = 0;
            threshold_timer.reset();
            threshold_interval = parallelChunks;
            threshold_draining = false;
          }
        }
 
        threshold_interval = threshold_interval > 0 ? threshold_interval - 1 : parallelChunks;
      }
 
      total_processed += pageInfo.GetLength();
      processed       += pageInfo.GetLength();
 
      st = dest->PutChunk( std::move( pageInfo ) );
      if( !st.IsOK() )
      {
        if( st.code == errRetry )
        {
          pResults->Set( "LastURL", dest->GetLastURL() );
          pResults->Set( "WrtRecoveryRedir", dest->GetWrtRecoveryRedir() );
          return SetResult( st );
        }
        return DestinationError( st );
      }
 
      if( progress )
      {
        progress->JobProgress( pJobId, total_processed, size );
        if( progress->ShouldCancel( pJobId ) )
          return SetResult( stError, errOperationInterrupted, kXR_Cancelled, "The copy-job has been cancelled!" );
      }
    }
 
    st = dest->Flush();
    if( !st.IsOK() )
      return DestinationError( st );
 
    //--------------------------------------------------------------------------
    // Copy extended attributes
    //--------------------------------------------------------------------------
    if( preserveXAttr && Utils::HasXAttr( GetSource() ) && Utils::HasXAttr( GetTarget() ) )
    {
      std::vector<xattr_t> xattrs;
      st = src->GetXAttr( xattrs );
      if( !st.IsOK() ) return SourceError( st );
      st = dest->SetXAttr( xattrs );
      if( !st.IsOK() ) return DestinationError( st );
    }
 
    //--------------------------------------------------------------------------
    // The size of the source is known and not enough data has been transferred
    // to the destination
    //--------------------------------------------------------------------------
    if( src->GetSize() >= 0 && size != total_processed )
    {
      log->Error( UtilityMsg, "The declared source size is %ld bytes, but "
                  "received %ld bytes.", size, total_processed );
      return SetResult( stError, errDataError );
    }
    pResults->Set( "size", total_processed );
 
    //--------------------------------------------------------------------------
    // Finalize the destination
    //--------------------------------------------------------------------------
    st = dest->Finalize();
    if( !st.IsOK() )
      return DestinationError( st );
 
    //--------------------------------------------------------------------------
    // Verify the checksums if needed
    //--------------------------------------------------------------------------
    if( checkSumMode != "none" )
    {
      log->Debug( UtilityMsg, "Attempting checksum calculation, mode: %s.",
                  checkSumMode.c_str() );
      std::string sourceCheckSum;
      std::string targetCheckSum;
 
      if( cptimer && cptimer->elapsed() > cpTimeout ) // check the CP timeout
        return SetResult( stError, errOperationExpired, 0, "CPTimeout exceeded." );
 
      //------------------------------------------------------------------------
      // Get the check sum at source
      //------------------------------------------------------------------------
      timeval oStart, oEnd;
      XRootDStatus st;
 
      if( checkSumMode == "end2end" || checkSumMode == "source" ||
          !checkSumPreset.empty() )
      {
        gettimeofday( &oStart, 0 );
        if( !checkSumPreset.empty() )
        {
          sourceCheckSum  = checkSumType + ":";
          sourceCheckSum += Utils::NormalizeChecksum( checkSumType,
                                                      checkSumPreset );
        }
        else
        {
          st = src->GetCheckSum( sourceCheckSum, checkSumType );
        }
        gettimeofday( &oEnd, 0 );
 
        if( !st.IsOK() )
          return SourceError( st );
 
        pResults->Set( "sourceCheckSum", sourceCheckSum );
      }
 
      if( !addcksums.empty() )
        pResults->Set( "additionalCkeckSum", src->GetAddCks() );
 
      if( cptimer && cptimer->elapsed() > cpTimeout ) // check the CP timeout
        return SetResult( stError, errOperationExpired, 0, "CPTimeout exceeded." );
 
      //------------------------------------------------------------------------
      // Get the check sum at destination
      //------------------------------------------------------------------------
      timeval tStart, tEnd;
 
      if( checkSumMode == "end2end" || checkSumMode == "target" )
      {
        gettimeofday( &tStart, 0 );
        st = dest->GetCheckSum( targetCheckSum, checkSumType );
        if( !st.IsOK() )
          return DestinationError( st );
        gettimeofday( &tEnd, 0 );
        pResults->Set( "targetCheckSum", targetCheckSum );
      }
 
      if( cptimer && cptimer->elapsed() > cpTimeout ) // check the CP timeout
        return SetResult( stError, errOperationExpired, 0, "CPTimeout exceeded." );
 
      //------------------------------------------------------------------------
      // Make sure the checksums are both lower case
      //------------------------------------------------------------------------
      auto sanitize_cksum = []( char c )
                            {
                              std::locale loc;
                              if( std::isalpha( c ) ) return std::tolower( c, loc );
                              return c;
                            };
 
      std::transform( sourceCheckSum.begin(), sourceCheckSum.end(),
                      sourceCheckSum.begin(), sanitize_cksum );
 
      std::transform( targetCheckSum.begin(), targetCheckSum.end(),
                      targetCheckSum.begin(), sanitize_cksum );
 
      //------------------------------------------------------------------------
      // Compare and inform monitoring
      //------------------------------------------------------------------------
      if( !sourceCheckSum.empty() && !targetCheckSum.empty() )
      {
        bool match = false;
        if( sourceCheckSum == targetCheckSum )
          match = true;
 
        Monitor *mon = DefaultEnv::GetMonitor();
        if( mon )
        {
          Monitor::CheckSumInfo i;
          i.transfer.origin = &GetSource();
          i.transfer.target = &GetTarget();
          i.cksum           = sourceCheckSum;
          i.oTime           = Utils::GetElapsedMicroSecs( oStart, oEnd );
          i.tTime           = Utils::GetElapsedMicroSecs( tStart, tEnd );
          i.isOK            = match;
          mon->Event( Monitor::EvCheckSum, &i );
        }
 
        if( !match )
        {
          if( rmOnBadCksum )
          {
            FileSystem fs( newDestUrl );
            st = fs.Rm( newDestUrl.GetPath() );
            if( !st.IsOK() )
              log->Error( UtilityMsg, "Invalid checksum: failed to remove the target file: %s", st.ToString().c_str() );
            else
              log->Info( UtilityMsg, "Target file removed due to bad checksum!" );
          }
 
          st = dest->Finalize();
          if( !st.IsOK() )
            log->Error( UtilityMsg, "Failed to finalize the destination: %s", st.ToString().c_str() );
 
          return SetResult( stError, errCheckSumError, 0 );
        }
 
        log->Info( UtilityMsg, "Checksum verification: succeeded." );
      }
    }
 
    return SetResult();
  }

References XrdCl::Monitor::CheckSumInfo::cksum, XrdCl::Status::code, XrdCl::Log::Debug(), XrdCl::errCheckSumError, XrdCl::errDataError, XrdCl::errInvalidArgs, XrdCl::errOperationExpired, XrdCl::errOperationInterrupted, XrdCl::Log::Error(), XrdCl::errRetry, XrdCl::errThresholdExceeded, XrdCl::Monitor::EvCheckSum, XrdCl::Monitor::Event(), XrdCl::PropertyList::Get(), XrdCl::Utils::GetElapsedMicroSecs(), XrdCl::PageInfo::GetLength(), XrdCl::DefaultEnv::GetLog(), XrdCl::DefaultEnv::GetMonitor(), XrdCl::URL::GetParams(), XrdCl::URL::GetPath(), XrdCl::CopyJob::GetSource(), XrdCl::CopyJob::GetTarget(), XrdCl::Utils::HasXAttr(), XrdCl::Utils::InferChecksumType(), XrdCl::Log::Info(), XrdCl::Monitor::CheckSumInfo::isOK, XrdCl::Status::IsOK(), XrdCl::CopyProgressHandler::JobProgress(), kXR_Cancelled, XrdCl::Utils::NormalizeChecksum(), XrdCl::Monitor::TransferInfo::origin, XrdCl::Monitor::CheckSumInfo::oTime, XrdCl::CopyJob::pJobId, XrdCl::CopyJob::pProperties, XrdCl::CopyJob::pResults, XrdCl::FileSystem::Rm(), XrdCl::PropertyList::Set(), XrdCl::URL::SetParams(), XrdCl::CopyProgressHandler::ShouldCancel(), sleep_nsec(), XrdCl::stError, XrdCl::suDone, XrdCl::Monitor::TransferInfo::target, time_nsec(), to_nsec(), XrdCl::Status::ToString(), XrdCl::Monitor::CheckSumInfo::transfer, XrdCl::Monitor::CheckSumInfo::tTime, XrdCl::UtilityMsg, and XrdCl::Log::Warning().

Here is the call graph for this function:

---

The documentation for this class was generated from the following files:

• XrdClClassicCopyJob.hh
• XrdClClassicCopyJob.cc