XrdCl::CopyProcess Class Reference

Copy the data from one point to another. More...

#include <XrdClCopyProcess.hh>

Collaboration diagram for XrdCl::CopyProcess:

Public Member Functions
	CopyProcess ()
	Constructor.
virtual	~CopyProcess ()
	Destructor.
XRootDStatus	AddJob (const PropertyList &properties, PropertyList *results)
XRootDStatus	Prepare ()
XRootDStatus	Run (CopyProgressHandler *handler)
	Run the copy jobs.

Detailed Description

Copy the data from one point to another.

Definition at line 107 of file XrdClCopyProcess.hh.

Constructor & Destructor Documentation

CopyProcess()

XrdCl::CopyProcess::CopyProcess

(

)

Constructor.

Definition at line 209 of file XrdClCopyProcess.cc.

                           : pImpl( new CopyProcessImpl() )
  {
  }

~CopyProcess()

XrdCl::CopyProcess::~CopyProcess ( )

virtual

Destructor.

Definition at line 216 of file XrdClCopyProcess.cc.

  {
    CleanUpJobs();
    delete pImpl;
  }

Member Function Documentation

AddJob()

XRootDStatus XrdCl::CopyProcess::AddJob	(	const PropertyList &	properties,
		PropertyList *	results )

Add job

Parameters

properties	job configuration parameters
results	placeholder for the results

Configuration properties: source [string] - original source URL target [string] - target directory or file sourceLimit [uint32_t] - maximum number sources force [bool] - overwrite target if exists posc [bool] - persistify only on successful close coerce [bool] - ignore locking semantics on destination makeDir [bool] - create path to the file if it doesn't exist thirdParty [string] - "first" try third party copy, if it fails try normal copy; "only" only try third party copy checkSumMode [string] - "none" - no checksumming "end2end" - end to end checksumming "source" - calculate checksum at source "target" - calculate checksum at target checkSumType [string] - type of the checksum to be used checkSumPreset [string] - checksum preset chunkSize [uint32_t] - size of a copy chunks in bytes parallelChunks [uint8_t] - number of chunks that should be requested in parallel initTimeout [time_t] - time limit for successfull initialization of the copy job tpcTimeout [time_t] - time limit for the actual copy to finish dynamicSource [bool] - support for the case where the size source file may change during reading process

Configuration job - this is a job that that is supposed to configure the copy process as a whole instead of adding a copy job:

jobType [string] - "configuration" - for configuraion parallel [uint8_t] - nomber of copy jobs to be run in parallel

Results: sourceCheckSum [string] - checksum at source, if requested targetCheckSum [string] - checksum at target, if requested size [uint64_t] - file size status [XRootDStatus] - status of the copy operation sources [vector<string>] - all sources used realTarget [string] - the actual disk server target

Definition at line 225 of file XrdClCopyProcess.cc.

  {
    Env *env = DefaultEnv::GetEnv();
 
    //--------------------------------------------------------------------------
    // Process a configuraion job
    //--------------------------------------------------------------------------
    if( properties.HasProperty( "jobType" ) &&
        properties.Get<std::string>( "jobType" ) == "configuration" )
    {
      if( pImpl->pJobProperties.size() > 0 &&
          pImpl->pJobProperties.rbegin()->HasProperty( "jobType" ) &&
          pImpl->pJobProperties.rbegin()->Get<std::string>( "jobType" ) == "configuration" )
      {
        PropertyList &config = *pImpl->pJobProperties.rbegin();
        PropertyList::PropertyMap::const_iterator it;
        for( it = properties.begin(); it != properties.end(); ++it )
          config.Set( it->first, it->second );
      }
      else
        pImpl->pJobProperties.push_back( properties );
      return XRootDStatus();
    }
 
    //--------------------------------------------------------------------------
    // Validate properties
    //--------------------------------------------------------------------------
    if( !properties.HasProperty( "source" ) )
      return XRootDStatus( stError, errInvalidArgs, 0, "source not specified" );
 
    if( !properties.HasProperty( "target" ) )
      return XRootDStatus( stError, errInvalidArgs, 0, "target not specified" );
 
    pImpl->pJobProperties.push_back( properties );
    PropertyList &p = pImpl->pJobProperties.back();
 
    const char *bools[] = {"target", "force", "posc", "coerce", "makeDir",
                           "zipArchive", "xcp", "preserveXAttr", "rmOnBadCksum",
                           "continue", "zipAppend", "doServer", 0};
    for( int i = 0; bools[i]; ++i )
      if( !p.HasProperty( bools[i] ) )
        p.Set( bools[i], false );
 
    if( !p.HasProperty( "thirdParty" ) )
      p.Set( "thirdParty", "none" );
 
    if( !p.HasProperty( "thirdPartyMode" ) )
      p.Set( "thirdPartyMode", "pull" );
 
    if( !p.HasProperty( "checkSumMode" ) )
      p.Set( "checkSumMode", "none" );
    else
    {
      if( !p.HasProperty( "checkSumType" ) )
      {
        pImpl->pJobProperties.pop_back();
        return XRootDStatus( stError, errInvalidArgs, 0,
                             "checkSumType not specified" );
      }
      else
      {
        //----------------------------------------------------------------------
        // Checksum type has to be case insensitive
        //----------------------------------------------------------------------
        std::string checkSumType;
        p.Get( "checkSumType", checkSumType );
        std::transform(checkSumType.begin(), checkSumType.end(),
                                                checkSumType.begin(), ::tolower);
        p.Set( "checkSumType", checkSumType );
      }
    }
 
    if( !p.HasProperty( "parallelChunks" ) )
    {
      int val = DefaultCPParallelChunks;
      env->GetInt( "CPParallelChunks", val );
      p.Set( "parallelChunks", val );
    }
 
    if( !p.HasProperty( "chunkSize" ) )
    {
      int val = DefaultCPChunkSize;
      env->GetInt( "CPChunkSize", val );
      p.Set( "chunkSize", val );
    }
 
    if( !p.HasProperty( "xcpBlockSize" ) )
    {
      int val = DefaultXCpBlockSize;
      env->GetInt( "XCpBlockSize", val );
      p.Set( "xcpBlockSize", val );
    }
 
    if( !p.HasProperty( "initTimeout" ) )
    {
      int val = DefaultCPInitTimeout;
      env->GetInt( "CPInitTimeout", val );
      p.Set( "initTimeout", val );
    }
 
    if( !p.HasProperty( "tpcTimeout" ) )
    {
      int val = DefaultCPTPCTimeout;
      env->GetInt( "CPTPCTimeout", val );
      p.Set( "tpcTimeout", val );
    }
 
    if( !p.HasProperty( "cpTimeout" ) )
    {
      int val = DefaultCPTimeout;
      env->GetInt( "CPTimeout", val );
      p.Set( "cpTimeout", val );
    }
 
    if( !p.HasProperty( "dynamicSource" ) )
      p.Set( "dynamicSource", false );
 
    if( !p.HasProperty( "xrate" ) )
      p.Set( "xrate", 0 );
 
    if( !p.HasProperty( "xrateThreshold" ) || p.Get<long long>( "xrateThreshold" ) == 0 )
    {
      int val = DefaultXRateThreshold;
      env->GetInt( "XRateThreshold", val );
      p.Set( "xrateThreshold", val );
    }
 
    //--------------------------------------------------------------------------
    // Insert the properties
    //--------------------------------------------------------------------------
    Log *log = DefaultEnv::GetLog();
    Utils::LogPropertyList( log, UtilityMsg, "Adding job with properties: %s",
                            p );
    pImpl->pJobResults.push_back( results );
    return XRootDStatus();
  }

References XrdCl::XRootDStatus::XRootDStatus(), XrdCl::PropertyList::begin(), XrdCl::DefaultCPChunkSize, XrdCl::DefaultCPInitTimeout, XrdCl::DefaultCPParallelChunks, XrdCl::DefaultCPTimeout, XrdCl::DefaultCPTPCTimeout, XrdCl::DefaultXCpBlockSize, XrdCl::DefaultXRateThreshold, XrdCl::PropertyList::end(), XrdCl::errInvalidArgs, XrdCl::PropertyList::Get(), XrdCl::DefaultEnv::GetEnv(), XrdCl::Env::GetInt(), XrdCl::DefaultEnv::GetLog(), XrdCl::PropertyList::HasProperty(), XrdCl::Utils::LogPropertyList(), XrdCl::PropertyList::Set(), XrdCl::stError, and XrdCl::UtilityMsg.

Referenced by DoCat(), and main().

Here is the call graph for this function:

Here is the caller graph for this function:

Prepare()

XRootDStatus XrdCl::CopyProcess::Prepare

(

)

Definition at line 366 of file XrdClCopyProcess.cc.

  {
    Log *log = DefaultEnv::GetLog();
    std::vector<PropertyList>::iterator it;
 
    log->Debug( UtilityMsg, "CopyProcess: %zu jobs to prepare",
                pImpl->pJobProperties.size() );
 
    std::map<std::string, uint32_t> targetFlags;
    int i = 0;
    for( it = pImpl->pJobProperties.begin(); it != pImpl->pJobProperties.end(); ++it, ++i )
    {
      PropertyList &props = *it;
 
      if( props.HasProperty( "jobType" ) &&
          props.Get<std::string>( "jobType" ) == "configuration" )
        continue;
 
      PropertyList *res   = pImpl->pJobResults[i];
      std::string tmp;
 
      props.Get( "source", tmp );
      URL source = tmp;
      if( !source.IsValid() )
      {
        log->Error( UtilityMsg, "Invalid copy source URL: %s", obfuscateAuth( tmp ).c_str() );
        return XRootDStatus( stError, errInvalidArgs, 0, "invalid source" );
      }
 
      //--------------------------------------------------------------------------
      // Create a virtual redirector if it is a Metalink file
      //--------------------------------------------------------------------------
      if( source.IsMetalink() )
      {
        RedirectorRegistry &registry = RedirectorRegistry::Instance();
        XRootDStatus st = registry.RegisterAndWait( source );
        if( !st.IsOK() ) return st;
      }
 
      // handle UNZIP CGI
      const URL::ParamsMap &cgi = source.GetParams();
      URL::ParamsMap::const_iterator itr = cgi.find( "xrdcl.unzip" );
      if( itr != cgi.end() )
      {
        props.Set( "zipArchive", true );
        props.Set( "zipSource",  itr->second );
      }
 
      props.Get( "target", tmp );
      URL target = tmp;
      if( !target.IsValid() )
      {
        log->Error( UtilityMsg, "Invalid copy target URL: %s", obfuscateAuth( tmp ).c_str() );
        return XRootDStatus( stError, errInvalidArgs, 0, "invalid target" );
      }
 
      if( target.GetProtocol() != "stdio" )
      {
        // handle directories
        bool targetIsDir = false;
        props.Get( "targetIsDir", targetIsDir );
 
        if( targetIsDir )
        {
          std::string path = target.GetPath() + '/';
          std::string fn;
 
          bool isZip = false;
          props.Get( "zipArchive", isZip );
          if( isZip )
          {
            props.Get( "zipSource", fn );
          }
          else if( source.IsMetalink() )
          {
            RedirectorRegistry &registry = XrdCl::RedirectorRegistry::Instance();
            VirtualRedirector *redirector = registry.Get( source );
            fn = redirector->GetTargetName();
          }
          else
          {
            fn = source.GetPath();
          }
 
          size_t pos = fn.rfind( '/' );
          if( pos != std::string::npos )
            fn = fn.substr( pos + 1 );
          path += fn;
          target.SetPath( path );
          props.Set( "target", target.GetURL() );
        }
      }
 
      bool tpc = false;
      props.Get( "thirdParty", tmp );
      if( tmp != "none" )
        tpc = true;
 
      //------------------------------------------------------------------------
      // Check if we have all we need
      //------------------------------------------------------------------------
      if( source.GetProtocol() != "stdio" && source.GetPath().empty() )
      {
        log->Debug( UtilityMsg, "CopyProcess (job #%d): no source specified.",
                    i );
        CleanUpJobs();
        XRootDStatus st = XRootDStatus( stError, errInvalidArgs );
        res->Set( "status", st );
        return st;
      }
 
      if( target.GetProtocol() != "stdio" && target.GetPath().empty() )
      {
        log->Debug( UtilityMsg, "CopyProcess (job #%d): no target specified.",
                    i );
        CleanUpJobs();
        XRootDStatus st = XRootDStatus( stError, errInvalidArgs );
        res->Set( "status", st );
        return st;
      }
 
      //------------------------------------------------------------------------
      // Check what kind of job we should do
      //------------------------------------------------------------------------
      CopyJob *job = 0;
 
      if( tpc == true )
      {
        MarkTPC( props );
        job = new TPFallBackCopyJob( i+1, &props, res );
      }
      else
        job = new ClassicCopyJob( i+1, &props, res );
 
      pImpl->pJobs.push_back( job );
    }
    return XRootDStatus();
  }

References XrdCl::ClassicCopyJob::ClassicCopyJob(), XrdCl::TPFallBackCopyJob::TPFallBackCopyJob(), XrdCl::XRootDStatus::XRootDStatus(), XrdCl::Log::Debug(), XrdCl::errInvalidArgs, XrdCl::Log::Error(), XrdCl::PropertyList::Get(), XrdCl::RedirectorRegistry::Get(), XrdCl::DefaultEnv::GetLog(), XrdCl::URL::GetParams(), XrdCl::URL::GetPath(), XrdCl::URL::GetProtocol(), XrdCl::VirtualRedirector::GetTargetName(), XrdCl::URL::GetURL(), XrdCl::PropertyList::HasProperty(), XrdCl::RedirectorRegistry::Instance(), XrdCl::URL::IsMetalink(), XrdCl::Status::IsOK(), XrdCl::URL::IsValid(), obfuscateAuth(), XrdCl::RedirectorRegistry::RegisterAndWait(), XrdCl::PropertyList::Set(), XrdCl::URL::SetPath(), XrdCl::stError, and XrdCl::UtilityMsg.

Referenced by DoCat(), and main().

Here is the call graph for this function:

Here is the caller graph for this function:

Run()

XRootDStatus XrdCl::CopyProcess::Run

(

CopyProgressHandler *

handler

)

Run the copy jobs.

Definition at line 508 of file XrdClCopyProcess.cc.

  {
    //--------------------------------------------------------------------------
    // Get the configuration
    //--------------------------------------------------------------------------
    uint8_t parallelThreads = 1;
    if( pImpl->pJobProperties.size() > 0 &&
        pImpl->pJobProperties.rbegin()->HasProperty( "jobType" ) &&
        pImpl->pJobProperties.rbegin()->Get<std::string>( "jobType" ) == "configuration" )
    {
      PropertyList &config = *pImpl->pJobProperties.rbegin();
      if( config.HasProperty( "parallel" ) )
        parallelThreads = (uint8_t)config.Get<int>( "parallel" );
    }
 
    //--------------------------------------------------------------------------
    // Run the show
    //--------------------------------------------------------------------------
    std::vector<CopyJob *>::iterator it;
    uint32_t currentJob = 1;
    uint32_t totalJobs  = pImpl->pJobs.size();
 
    //--------------------------------------------------------------------------
    // Single thread
    //--------------------------------------------------------------------------
    if( parallelThreads == 1 )
    {
      XRootDStatus err;
 
      for( it = pImpl->pJobs.begin(); it != pImpl->pJobs.end(); ++it )
      {
        QueuedCopyJob j( *it, progress, currentJob, totalJobs );
        j.Run(0);
 
        XRootDStatus st = (*it)->GetResults()->Get<XRootDStatus>( "status" );
        if( err.IsOK() && !st.IsOK() )
        {
          err = st;
        }
        ++currentJob;
      }
 
      if( !err.IsOK() ) return err;
    }
    //--------------------------------------------------------------------------
    // Multiple threads
    //--------------------------------------------------------------------------
    else
    {
      uint32_t workers = std::min( (uint32_t)parallelThreads,
                                   (uint32_t)pImpl->pJobs.size() );
      JobManager jm( workers );
      jm.Initialize();
      if( !jm.Start() )
        return XRootDStatus( stError, errOSError, 0,
                             "Unable to start job manager" );
 
      XrdSysSemaphore *sem = new XrdSysSemaphore(0);
      std::vector<QueuedCopyJob*> queued;
      for( it = pImpl->pJobs.begin(); it != pImpl->pJobs.end(); ++it )
      {
        QueuedCopyJob *j = new QueuedCopyJob( *it, progress, currentJob,
                                              totalJobs, sem );
 
        queued.push_back( j );
        jm.QueueJob(j, 0);
        ++currentJob;
      }
 
      std::vector<QueuedCopyJob*>::iterator itQ;
      for( itQ = queued.begin(); itQ != queued.end(); ++itQ )
        sem->Wait();
      delete sem;
 
      if( !jm.Stop() )
        return XRootDStatus( stError, errOSError, 0,
                             "Unable to stop job manager" );
      jm.Finalize();
      for( itQ = queued.begin(); itQ != queued.end(); ++itQ )
        delete *itQ;
 
      for( it = pImpl->pJobs.begin(); it != pImpl->pJobs.end(); ++it )
      {
        XRootDStatus st = (*it)->GetResults()->Get<XRootDStatus>( "status" );
        if( !st.IsOK() ) return st;
      }
    };
    return XRootDStatus();
  }

References XrdSysSemaphore::XrdSysSemaphore(), XrdCl::XRootDStatus::XRootDStatus(), XrdCl::errOSError, XrdCl::JobManager::Finalize(), XrdCl::PropertyList::Get(), XrdCl::PropertyList::HasProperty(), XrdCl::JobManager::Initialize(), XrdCl::Status::IsOK(), XrdCl::JobManager::QueueJob(), XrdCl::JobManager::Start(), XrdCl::stError, XrdCl::JobManager::Stop(), and XrdSysSemaphore::Wait().

Referenced by DoCat(), and main().

Here is the call graph for this function:

Here is the caller graph for this function:

---

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

• XrdClCopyProcess.hh
• XrdClCopyProcess.cc