XrdClCopy.cc

Go to the documentation of this file.

//------------------------------------------------------------------------------
// Copyright (c) 2011-2014 by European Organization for Nuclear Research (CERN)
// Author: Lukasz Janyst <ljanyst@cern.ch>
//------------------------------------------------------------------------------
// This file is part of the XRootD software suite.
//
// 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/>.
//
// In applying this licence, CERN does not waive the privileges and immunities
// granted to it by virtue of its status as an Intergovernmental Organization
// or submit itself to any jurisdiction.
//------------------------------------------------------------------------------
 
#include "XrdApps/XrdCpConfig.hh"
#include "XrdApps/XrdCpFile.hh"
#include "XrdCl/XrdClConstants.hh"
#include "XrdCl/XrdClCopyProcess.hh"
#include "XrdCl/XrdClDefaultEnv.hh"
#include "XrdCl/XrdClLog.hh"
#include "XrdCl/XrdClFileSystem.hh"
#include "XrdCl/XrdClUtils.hh"
#include "XrdCl/XrdClDlgEnv.hh"
#include "XrdCl/XrdClOptimizers.hh"
#include "XrdSys/XrdSysE2T.hh"
#include "XrdSys/XrdSysPthread.hh"
#include "XrdOuc/XrdOucPrivateUtils.hh"
 
#include <cstdio>
#include <iostream>
#include <iomanip>
#include <limits>
#include <string>
 
using namespace std::string_literals;
 
//------------------------------------------------------------------------------
// Progress notifier
//------------------------------------------------------------------------------
class ProgressDisplay: public XrdCl::CopyProgressHandler
{
  public:
    //--------------------------------------------------------------------------
    //--------------------------------------------------------------------------
    ProgressDisplay(): pPrevious(0), pPrintProgressBar(true),
      pPrintSourceCheckSum(false), pPrintTargetCheckSum(false),
      pPrintAdditionalCheckSum(false)
    {}
 
    //--------------------------------------------------------------------------
    //--------------------------------------------------------------------------
    virtual void BeginJob( uint32_t          jobNum,
                           uint32_t          jobTotal,
                           const XrdCl::URL *source,
                           const XrdCl::URL *destination )
    {
      XrdSysMutexHelper scopedLock( pMutex );
      if( pPrintProgressBar )
      {
        if( jobTotal > 1 )
        {
          std::cerr << "Job: "    << jobNum << "/" << jobTotal << std::endl;
          std::cerr << "Source: " << source->GetURL() << std::endl;
          std::cerr << "Target: " << destination->GetURL() << std::endl;
        }
      }
      pPrevious = 0;
 
      JobData d;
      d.started = time(0);
      d.source  = source;
      d.target  = destination;
      pOngoingJobs[jobNum] = d;
    }
 
    //--------------------------------------------------------------------------
    //--------------------------------------------------------------------------
    virtual void EndJob( uint32_t jobNum, const XrdCl::PropertyList *results )
    {
      XrdSysMutexHelper scopedLock( pMutex );
 
      std::map<uint32_t, JobData>::iterator it = pOngoingJobs.find( jobNum );
      if( it == pOngoingJobs.end() )
        return;
 
      JobData &d = it->second;
 
      // make sure the last available status was printed, which may not be
      // the case when processing stdio since we throttle printing and don't
      // know the total size
      JobProgress( jobNum, d.bytesProcessed, d.bytesTotal );
 
      if( pPrintProgressBar )
      {
        if( pOngoingJobs.size() > 1 )
          std::cerr << "\r" << std::string(70, ' ') << "\r";
        else
          std::cerr << std::endl;
      }
 
      XrdCl::XRootDStatus st;
      results->Get( "status", st );
      if( !st.IsOK() )
      {
        pOngoingJobs.erase(it);
        return;
      }
 
      std::string checkSum;
      uint64_t    size;
      results->Get( "size", size );
      if( pPrintSourceCheckSum )
      {
        results->Get( "sourceCheckSum", checkSum );
        PrintCheckSum( d.source, checkSum, size );
      }
 
      if( pPrintTargetCheckSum )
      {
        results->Get( "targetCheckSum", checkSum );
        PrintCheckSum( d.target, checkSum, size );
      }
 
      if( pPrintAdditionalCheckSum )
      {
        std::vector<std::string> addcksums;
        results->Get( "additionalCkeckSum", addcksums );
        for( auto &cks : addcksums )
          PrintCheckSum( d.source, cks, size );
      }
 
      pOngoingJobs.erase(it);
    }
 
    //--------------------------------------------------------------------------
    //--------------------------------------------------------------------------
    std::string GetProgressBar( time_t now )
    {
      JobData &d = pOngoingJobs.begin()->second;
 
      uint64_t speed = 0;
      if( now-d.started )
        speed = d.bytesProcessed/(now-d.started);
      else
        speed = d.bytesProcessed;
 
      std::string bar;
      int prog = 0;
      int proc = 0;
 
      if( d.bytesTotal )
      {
        prog = (int)((double)d.bytesProcessed/d.bytesTotal*50);
        proc = (int)((double)d.bytesProcessed/d.bytesTotal*100);
      }
      else
      {
        prog = 50;
        proc = 100;
      }
      bar.append( prog, '=' );
      if( prog < 50 )
        bar += ">";
 
      std::ostringstream o;
      o << "[" << XrdCl::Utils::BytesToString(d.bytesProcessed) << "B/";
      o << XrdCl::Utils::BytesToString(d.bytesTotal) << "B]";
      o << "[" << std::setw(3) << std::right << proc << "%]";
      o << "[" << std::setw(50) << std::left;
      o << bar;
      o << "]";
      o << "[" << XrdCl::Utils::BytesToString(speed) << "B/s]  ";
      return o.str();
    }
 
    //--------------------------------------------------------------------------
    //--------------------------------------------------------------------------
    std::string GetSummaryBar( time_t now )
    {
      std::map<uint32_t, JobData>::iterator it;
      std::ostringstream o;
 
      for( it = pOngoingJobs.begin(); it != pOngoingJobs.end(); ++it )
      {
        JobData  &d      = it->second;
        uint32_t  jobNum = it->first;
 
        uint64_t speed = 0;
        if( now-d.started )
          speed = d.bytesProcessed/(now-d.started);
 
        int proc = 0;
        if( d.bytesTotal )
          proc = (int)((double)d.bytesProcessed/d.bytesTotal*100);
        else
          proc = 100;
 
        o << "[#" << jobNum << ": ";
        o << proc << "% ";
        o << XrdCl::Utils::BytesToString(speed) << "B/s] ";
      }
      o << "        ";
      return o.str();
    }
 
    //--------------------------------------------------------------------------
    //--------------------------------------------------------------------------
    virtual void JobProgress( uint32_t jobNum,
                              uint64_t bytesProcessed,
                              uint64_t bytesTotal )
    {
      XrdSysMutexHelper scopedLock( pMutex );
 
      if( pPrintProgressBar )
      {
        time_t now = time(0);
        if( (now - pPrevious < 1) && (bytesProcessed != bytesTotal) )
          return;
        pPrevious = now;
 
        std::map<uint32_t, JobData>::iterator it = pOngoingJobs.find( jobNum );
        if( it == pOngoingJobs.end() )
          return;
 
        JobData &d = it->second;
 
        d.bytesProcessed = bytesProcessed;
        d.bytesTotal     = bytesTotal;
 
        std::string progress;
        if( pOngoingJobs.size() == 1 )
          progress = GetProgressBar( now );
        else
          progress = GetSummaryBar( now );
 
        std::cerr << "\r" << progress << std::flush;
      }
    }
 
    //--------------------------------------------------------------------------
    //--------------------------------------------------------------------------
    void PrintCheckSum( const XrdCl::URL  *url,
                        const std::string &checkSum,
                        uint64_t size )
    {
      if( checkSum.empty() )
        return;
      std::string::size_type i = checkSum.find( ':' );
      std::string type  = checkSum.substr( 0, i );
      std::string value = checkSum.substr( i+1, checkSum.length()-i );
 
      //------------------------------------------------------------------------
      // The comparison uses adler32 and crc32 without leading zeros (see
      // Utils::NormalizeChecksum), but print all 8 digits, as xrdadler32 and
      // the servers do
      //------------------------------------------------------------------------
      if( ( type == "adler32" || type == "crc32" ) && value.length() < 8 )
        value.insert( 0, 8 - value.length(), '0' );
 
      std::cerr << type << ": " << value << " ";
 
      if( url->IsLocalFile() )
        std::cerr << url->GetPath() << " ";
      else
      {
        std::cerr << url->GetProtocol() << "://" << url->GetHostId();
        std::cerr << url->GetPath() << " ";
      }
 
      std::cerr << size;
      std::cerr << std::endl;
    }
 
    //--------------------------------------------------------------------------
    // Printing flags
    //--------------------------------------------------------------------------
    void PrintProgressBar( bool print )    { pPrintProgressBar    = print; }
    void PrintSourceCheckSum( bool print ) { pPrintSourceCheckSum = print; }
    void PrintTargetCheckSum( bool print ) { pPrintTargetCheckSum = print; }
    void PrintAdditionalCheckSum( bool print ) { pPrintAdditionalCheckSum = print; }
 
  private:
    struct JobData
    {
      JobData(): bytesProcessed(0), bytesTotal(0),
        started(0), source(0), target(0) {}
      uint64_t          bytesProcessed;
      uint64_t          bytesTotal;
      time_t            started;
      const XrdCl::URL *source;
      const XrdCl::URL *target;
    };
 
    time_t                      pPrevious;
    bool                        pPrintProgressBar;
    bool                        pPrintSourceCheckSum;
    bool                        pPrintTargetCheckSum;
    bool                        pPrintAdditionalCheckSum;
    std::map<uint32_t, JobData> pOngoingJobs;
    XrdSysRecMutex              pMutex;
};
 
//------------------------------------------------------------------------------
// Check if we support all the specified user options
//------------------------------------------------------------------------------
bool AllOptionsSupported( XrdCpConfig *config )
{
  if( config->pHost )
  {
    std::cerr << "SOCKS Proxies are not yet supported" << std::endl;
    return false;
  }
 
  return true;
}
 
//------------------------------------------------------------------------------
// Append extra cgi info to existing URL
//------------------------------------------------------------------------------
void AppendCGI( std::string &url, const char *newCGI )
{
  if( !newCGI || !(*newCGI) )
    return;
 
  if( *newCGI == '&' )
    ++newCGI;
 
  if( url.find( '?' ) == std::string::npos )
    url += "?";
 
  if( url.find( '&' ) == std::string::npos )
    url += "&";
 
  url += newCGI;
}
 
//------------------------------------------------------------------------------
// Process commandline environment settings
//------------------------------------------------------------------------------
void ProcessCommandLineEnv( XrdCpConfig *config )
{
  XrdCl::Env *env = XrdCl::DefaultEnv::GetEnv();
 
  XrdCpConfig::defVar *cursor = config->intDefs;
  while( cursor )
  {
    env->PutInt( cursor->vName, cursor->intVal );
    cursor = cursor->Next;
  }
 
  cursor = config->strDefs;
  while( cursor )
  {
    env->PutString( cursor->vName, cursor->strVal );
    cursor = cursor->Next;
  }
}
 
//------------------------------------------------------------------------------
// Translate file type to a string for diagnostics purposes
//------------------------------------------------------------------------------
const char *FileType2String( XrdCpFile::PType type )
{
  switch( type )
  {
    case XrdCpFile::isDir:   return "directory";
    case XrdCpFile::isFile:  return "local file";
    case XrdCpFile::isXroot: return "xroot";
    case XrdCpFile::isXroots: return "xroots";
    case XrdCpFile::isHttp:  return "http";
    case XrdCpFile::isHttps: return "https";
    case XrdCpFile::isStdIO: return "stdio";
    default: return "other";
  };
}
 
//------------------------------------------------------------------------------
// Count the sources
//------------------------------------------------------------------------------
uint32_t CountSources( XrdCpFile *file )
{
  uint32_t count;
  for( count = 0; file; file = file->Next, ++count ) {};
  return count;
}
 
//------------------------------------------------------------------------------
// Adjust file information for the cases when XrdCpConfig cannot do this
//------------------------------------------------------------------------------
void AdjustFileInfo( XrdCpFile *file )
{
  //----------------------------------------------------------------------------
  // If the file is url and the directory offset is not set we set it
  // to the last slash
  //----------------------------------------------------------------------------
  if( file->Doff == 0 )
  {
    char *slash = file->Path;
    for( ; *slash; ++slash ) {};
    for( ; *slash != '/' && slash > file->Path; --slash ) {};
    file->Doff = slash - file->Path;
  }
};
 
//------------------------------------------------------------------------------
// Recursively index all files and directories inside a remote directory
//------------------------------------------------------------------------------
XrdCpFile *IndexRemote( XrdCl::FileSystem *fs,
                        std::string        basePath,
                        long               dirOffset )
{
  using namespace XrdCl;
 
  Log *log = DefaultEnv::GetLog();
  log->Debug( AppMsg, "Indexing %s", basePath.c_str() );
 
  DirectoryList *dirList = 0;
  XRootDStatus st = fs->DirList( URL( basePath ).GetPath(), DirListFlags::Recursive
      | DirListFlags::Locate | DirListFlags::Merge, dirList );
  if( !st.IsOK() )
  {
    log->Info( AppMsg, "Failed to get directory listing for %s: %s",
                       basePath.c_str(),
                       st.GetErrorMessage().c_str() );
    return 0;
  }
 
  XrdCpFile start, *current = 0;
  XrdCpFile *end   = &start;
  int       badUrl = 0;
  for( auto itr = dirList->Begin(); itr != dirList->End(); ++itr )
  {
    DirectoryList::ListEntry *e = *itr;
    if( e->GetStatInfo()->TestFlags( StatInfo::IsDir ) )
      continue;
    std::string path = basePath + '/' + e->GetName();
    current = new XrdCpFile( path.c_str(), badUrl );
    if( badUrl )
    {
      log->Error( AppMsg, "Bad URL: %s", current->Path );
      delete current;
      return 0;
    }
 
    current->Doff = dirOffset;
    end->Next     = current;
    end           = current;
  }
 
  delete dirList;
 
  return start.Next;
}
 
//------------------------------------------------------------------------------
// Clean up the copy job descriptors
//------------------------------------------------------------------------------
void CleanUpResults( std::vector<XrdCl::PropertyList *> &results )
{
  std::vector<XrdCl::PropertyList *>::iterator it;
  for( it = results.begin(); it != results.end(); ++it )
    delete *it;
}
 
//--------------------------------------------------------------------------
// Let the show begin
//------------------------------------------------------------------------------
int main( int argc, char **argv )
{
  using namespace XrdCl;
 
  //----------------------------------------------------------------------------
  // Configure the copy command, if it returns then everything went well, ugly
  //----------------------------------------------------------------------------
  XrdCpConfig config( argv[0] );
  config.Config( argc, argv, XrdCpConfig::optRmtRec );
  if( !AllOptionsSupported( &config ) )
    return 50; // generic error
  ProcessCommandLineEnv( &config );
 
  //----------------------------------------------------------------------------
  // Set options
  //----------------------------------------------------------------------------
  CopyProcess process;
  Log *log = DefaultEnv::GetLog();
  if( config.Dlvl )
  {
    if( config.Dlvl == 1 ) log->SetLevel( Log::InfoMsg );
    else if( config.Dlvl == 2 ) log->SetLevel( Log::DebugMsg );
    else if( config.Dlvl == 3 ) log->SetLevel( Log::DumpMsg );
  }
 
  ProgressDisplay progress;
  if( config.Want(XrdCpConfig::DoNoPbar) || !isatty( fileno( stdout ) ) )
    progress.PrintProgressBar( false );
 
  bool         posc          = false;
  bool         force         = false;
  bool         coerce        = false;
  bool         makedir       = false;
  bool         dynSrc        = false;
  bool         delegate      = false;
  bool         preserveXAttr = false;
  bool         rmOnBadCksum  = false;
  bool         continue_     = false;
  bool         zipappend     = false;
  bool         doserver      = false;
  std::string thirdParty     = "none"s;
  std::string thirdPartyMode = "pull"s;
 
  const std::string thirdPartyTokenFile = config.tcpTokenFile ? config.tcpTokenFile : ""s;
 
  if( config.Want( XrdCpConfig::DoPosc ) )        posc           = true;
  if( config.Want( XrdCpConfig::DoForce ) )       force          = true;
  if( config.Want( XrdCpConfig::DoCoerce ) )      coerce         = true;
  if( config.Want( XrdCpConfig::DoTpc ) )         thirdParty     = "first"s;
  if( config.Want( XrdCpConfig::DoTpcOnly ) )     thirdParty     = "only"s;
  if( config.Want( XrdCpConfig::DoTpcModePush ) ) thirdPartyMode = "push"s;
  if( config.Want( XrdCpConfig::DoZipAppend ) ) zipappend  = true;
  if( config.Want( XrdCpConfig::DoServer ) )    doserver   = true;
  if( config.Want( XrdCpConfig::DoTpcDlgt ) )
  {
    // the env var is being set already here (we are issuing a stat
    // inhere and we need the env var when we are establishing the
    // connection and authenticating), but we are also setting a delegate
    // parameter for CopyJob so it can be used on its own.
    DlgEnv::Instance().Enable();
    delegate = true;
  }
  else
    DlgEnv::Instance().Disable();
 
  if( config.Want( XrdCpConfig::DoRecurse ) )     makedir       = true;
  if( config.Want( XrdCpConfig::DoPath    ) )      makedir       = true;
  if( config.Want( XrdCpConfig::DoDynaSrc ) )      dynSrc        = true;
  if( config.Want( XrdCpConfig::DoXAttr ) )        preserveXAttr = true;
  if( config.Want( XrdCpConfig::DoRmOnBadCksum ) ) rmOnBadCksum  = true;
  if( config.Want( XrdCpConfig::DoContinue ) )     continue_     = true;
 
  if( force && continue_ )
  {
    std::cerr << "Invalid argument combination: continue + force." << std::endl;
    return 50;
  }
 
  //----------------------------------------------------------------------------
  // Checksums
  //----------------------------------------------------------------------------
  std::string checkSumType;
  std::string checkSumPreset;
  std::string checkSumMode  = "none";
  if( config.Want( XrdCpConfig::DoCksum ) )
  {
    checkSumMode = "end2end";
    std::vector<std::string> ckSumParams;
    Utils::splitString( ckSumParams, config.CksVal, ":" );
    if( ckSumParams.size() > 1 )
    {
      if( ckSumParams[1] == "print" )
      {
        checkSumMode = "target";
        progress.PrintTargetCheckSum( true );
      }
      else
        checkSumPreset = ckSumParams[1];
    }
    checkSumType = ckSumParams[0];
  }
 
  if( config.Want( XrdCpConfig::DoCksrc ) )
  {
    checkSumMode = "source";
    std::vector<std::string> ckSumParams;
    Utils::splitString( ckSumParams, config.CksVal, ":" );
    if( ckSumParams.size() == 2 )
    {
      checkSumMode = "source";
      checkSumType = ckSumParams[0];
      progress.PrintSourceCheckSum( true );
    }
    else
    {
      std::cerr << "Invalid parameter: " << config.CksVal << std::endl;
      return 50; // generic error
    }
  }
 
  if( !config.AddCksVal.empty() )
    progress.PrintAdditionalCheckSum( true );
 
  //----------------------------------------------------------------------------
  // ZIP archive
  //----------------------------------------------------------------------------
  std::string zipFile;
  bool        zip = false;
  if( config.Want( XrdCpConfig::DoZip ) )
  {
    zipFile = config.zipFile;
    zip = true;
  }
 
  //----------------------------------------------------------------------------
  // Extreme Copy
  //----------------------------------------------------------------------------
  int nbSources = 0;
  bool xcp      = false;
  if( config.Want( XrdCpConfig::DoSources ) )
  {
    nbSources = config.nSrcs;
    xcp       = true;
  }
 
  //----------------------------------------------------------------------------
  // Environment settings
  //----------------------------------------------------------------------------
  XrdCl::Env *env = XrdCl::DefaultEnv::GetEnv();
 
  /* Stop PostMaster when exiting main() to ensure proper shutdown */
  struct scope_exit {
    ~scope_exit() { XrdCl::DefaultEnv::GetPostMaster()->Stop(); }
  } stopPostMaster;
 
  if( config.nStrm != 0 )
    env->PutInt( "SubStreamsPerChannel", config.nStrm + 1 /*stands for the control stream*/ );
 
  if( config.Retry != -1 )
  {
    env->PutInt( "CpRetry", config.Retry );
    env->PutString( "CpRetryPolicy", config.RetryPolicy );
  }
 
  if( config.Want( XrdCpConfig::DoNoTlsOK ) )
    env->PutInt( "NoTlsOK", 1 );
 
  if( config.Want( XrdCpConfig::DoTlsNoData ) )
    env->PutInt( "TlsNoData", 1 );
 
  if( config.Want( XrdCpConfig::DoTlsMLF ) )
    env->PutInt( "TlsMetalink", 1 );
 
  if( config.Want( XrdCpConfig::DoZipMtlnCksum ) )
    env->PutInt( "ZipMtlnCksum", 1 );
 
  //----------------------------------------------------------------------------
  // Headers to inject into the HTTP requests. These are handed to the XrdClHttp
  // plug-in verbatim, as a newline separated list; that plug-in owns all of the
  // parsing and validation of their contents.
  //----------------------------------------------------------------------------
  if( !config.HttpHeaders.empty() )
  {
    std::string headers;
    for( auto &header : config.HttpHeaders )
    {
      if( !headers.empty() ) headers += '\n';
      headers += header;
    }
    env->PutString( "HttpHeaders", headers );
  }
 
  int chunkSize = DefaultCPChunkSize;
  env->GetInt( "CPChunkSize", chunkSize );
 
  int blockSize = DefaultXCpBlockSize;
  env->GetInt( "XCpBlockSize", blockSize );
 
  int parallelChunks = DefaultCPParallelChunks;
  env->GetInt( "CPParallelChunks", parallelChunks );
  if( parallelChunks < 1 ||
      parallelChunks > std::numeric_limits<uint8_t>::max() )
  {
    std::cerr << "Can only handle between 1 and ";
    std::cerr << (int)std::numeric_limits<uint8_t>::max();
    std::cerr << " chunks in parallel. You asked for " << parallelChunks;
    std::cerr << "." << std::endl;
    return 50; // generic error
  }
 
  if( !preserveXAttr )
  {
    int val = DefaultPreserveXAttrs;
    env->GetInt( "PreserveXAttrs", val );
    if( val ) preserveXAttr = true;
  }
 
  log->Dump( AppMsg, "Chunk size: %d, parallel chunks %d, streams: %d",
             chunkSize, parallelChunks, config.nStrm + 1 );
 
  //----------------------------------------------------------------------------
  // Build the URLs
  //----------------------------------------------------------------------------
  std::vector<XrdCl::PropertyList*> resultVect;
 
  std::string dest;
  if( config.dstFile->Protocol == XrdCpFile::isDir ||
      config.dstFile->Protocol == XrdCpFile::isFile )
  {
    dest = "file://";
 
    // if it is not an absolute path append cwd
    if( config.dstFile->Path[0] != '/' )
    {
      char buf[FILENAME_MAX];
      char *cwd = getcwd( buf, FILENAME_MAX );
      if( !cwd )
      {
        XRootDStatus st( stError, XProtocol::mapError( errno ), errno );
        std::cerr <<  st.GetErrorMessage() << std::endl;
        return st.GetShellCode();
      }
      dest += cwd;
      dest += '/';
    }
  }
  dest += config.dstFile->Path;
 
  //----------------------------------------------------------------------------
  // A destination that ends in / is a directory. Several sources are allowed,
  // and the source file name is added. root://host/ loses that slash when the
  // address is parsed, so the slash is read from the text the user typed.
  //----------------------------------------------------------------------------
  bool targetIsDir = config.dstFile->Protocol == XrdCpFile::isDir;
  if( !targetIsDir &&
      ( config.dstFile->Protocol == XrdCpFile::isXroot ||
        config.dstFile->Protocol == XrdCpFile::isXroots ||
        config.dstFile->Protocol == XrdCpFile::isHttp ||
        config.dstFile->Protocol == XrdCpFile::isHttps ||
        config.dstFile->Protocol == XrdCpFile::isS3 ) )
  {
    URL target( dest );
    const std::string path = target.GetPath();
    if( !path.empty() )
      targetIsDir = path.back() == '/';
    else
    {
      const std::string spec = config.dstFile->Path;
      const size_t query = spec.find( '?' );
      const size_t end = query == std::string::npos ? spec.size() : query;
      targetIsDir = end > 0 && spec[end - 1] == '/';
    }
  }
 
  // A local directory has already been replaced by the files inside it.
  // The destination of that copy is a directory.
  if( !targetIsDir && config.Want( XrdCpConfig::DoRecurse ) )
  {
    for( XrdCpFile *file = config.srcFile; file; file = file->Next )
    {
      if( strncmp( file->ProtName, "file", 4 ) == 0 &&
          std::string( file->Path ).size() == size_t( file->Doff + file->Dlen ) )
      {
        targetIsDir = true;
        break;
      }
    }
  }
 
  // root:// and roots:// may already be a directory even when the path does
  // not end in /. Ask the server, and add the source file name if it is.
  // http://, https://, and s3:// are directories only when the path ends in /.
  if( !targetIsDir &&
      ( config.dstFile->Protocol == XrdCpFile::isXroot ||
        config.dstFile->Protocol == XrdCpFile::isXroots ) )
  {
    URL target( dest );
    FileSystem fs( target );
    StatInfo *statInfo = 0;
    XRootDStatus st = fs.Stat( target.GetPathWithParams(), statInfo );
    if( st.IsOK() && statInfo && statInfo->TestFlags( StatInfo::IsDir ) )
      targetIsDir = true;
    else if( st.errNo == kXR_NotAuthorized )
    {
      log->Error( AppMsg, "%s (destination)", st.ToString().c_str() );
      std::cerr << st.ToStr() << std::endl;
      delete statInfo;
      return st.GetShellCode();
    }
    delete statInfo;
  }
 
  //----------------------------------------------------------------------------
  // If we have multiple sources and target is neither a directory nor stdout
  // then we cannot proceed
  //----------------------------------------------------------------------------
  if( CountSources(config.srcFile) > 1 && !targetIsDir &&
      config.dstFile->Protocol != XrdCpFile::isStdIO )
  {
    std::cerr << "Multiple sources were given but target is not a directory.";
    std::cerr << std::endl;
    return 50; // generic error
  }
 
  //----------------------------------------------------------------------------
  // If we're doing remote recursive copy, chain all the files (if it's a
  // directory)
  //----------------------------------------------------------------------------
  bool remoteSrcIsDir = false;
  if( config.Want( XrdCpConfig::DoRecurse ) &&
      (config.srcFile->Protocol == XrdCpFile::isXroot ||
       config.srcFile->Protocol == XrdCpFile::isXroots) )
  {
    URL          source( config.srcFile->Path );
    FileSystem  *fs       = new FileSystem( source );
    StatInfo    *statInfo = 0;
 
    XRootDStatus st = fs->Stat( source.GetPath(), statInfo );
    if( st.IsOK() && statInfo->TestFlags( StatInfo::IsDir ) )
    {
      remoteSrcIsDir = true;
      // Each listed file name is added to this destination.
      targetIsDir = true;
      //------------------------------------------------------------------------
      // Recursively index the remote directory
      //------------------------------------------------------------------------
      delete config.srcFile;
      std::string url = source.GetURL();
      config.srcFile = IndexRemote( fs, url, url.size() );
      if ( !config.srcFile )
      {
        std::cerr << "Error indexing remote directory.";
        return 50; // generic error
      }
    }
 
    delete fs;
    delete statInfo;
  }
 
  XrdCpFile *sourceFile = config.srcFile;
  //----------------------------------------------------------------------------
  // Process the sources
  //----------------------------------------------------------------------------
  while( sourceFile )
  {
    AdjustFileInfo( sourceFile );
 
    //--------------------------------------------------------------------------
    // Create a job for every source
    //--------------------------------------------------------------------------
    PropertyList  properties;
    PropertyList *results = new PropertyList;
    std::string source = sourceFile->Path;
    if( sourceFile->Protocol == XrdCpFile::isFile )
    {
      // make sure it is an absolute path
      if( source[0] == '/' )
        source = "file://" + source;
      else
      {
        char buf[FILENAME_MAX];
        char *cwd = getcwd( buf, FILENAME_MAX );
        if( !cwd )
        {
          XRootDStatus st( stError, XProtocol::mapError( errno ), errno );
          std::cerr <<  st.GetErrorMessage() << std::endl;
          return st.GetShellCode();
        }
        source = "file://" + std::string( cwd ) + '/' + source;
      }
    }
 
    AppendCGI( source, config.srcOpq );
 
    std::string sourcePathObf = sourceFile->Path;
    std::string destPathObf = dest;
    if( unlikely(log->GetLevel() >= Log::DumpMsg) ) {
      sourcePathObf = obfuscateAuth(sourcePathObf);
      destPathObf = obfuscateAuth(destPathObf);
    }
    log->Dump( AppMsg, "Processing source entry: %s, type %s, target file: %s, logLevel = %d",
               sourcePathObf.c_str(), FileType2String( sourceFile->Protocol ),
               destPathObf.c_str(), log->GetLevel() );
 
    //--------------------------------------------------------------------------
    // Set up the job
    //--------------------------------------------------------------------------
    std::string target = dest;
 
 
    bool srcIsDir = false;
    // if this is local file, for a directory Dlen + Doff will overlap with path size
    if( strncmp( sourceFile->ProtName, "file", 4 ) == 0 )
      srcIsDir = std::string( sourceFile->Path ).size() == size_t( sourceFile->Doff + sourceFile->Dlen );
    // otherwise we are handling a remote file
    else
      srcIsDir = remoteSrcIsDir;
    // if this is a recursive copy make sure we preserve the directory structure
    if( config.Want( XrdCpConfig::DoRecurse ) && srcIsDir )
    {
      // get the source directory
      std::string srcDir( sourceFile->Path, sourceFile->Doff );
      // remove the trailing slash
      if( !srcDir.empty() && srcDir.back() == '/' )
        srcDir.pop_back();
      size_t diroff = srcDir.rfind( '/' );
      // A relative name has no slash, so the whole name is kept. Otherwise move
      // past the slash in front of the directory name. Keeping it would write
      // two slashes into the destination.
      if( diroff == std::string::npos )
        diroff = 0;
      else
        ++diroff;
      if( target.empty() || target.back() != '/' )
        target += '/';
      target += sourceFile->Path + diroff;
      // remove the filename from destination path as it will be appended later anyway
      target = target.substr( 0 , target.rfind('/') );
    }
    AppendCGI( target, config.dstOpq );
 
    properties.Set( "source",          source                 );
    properties.Set( "target",          target                 );
    properties.Set( "force",           force                  );
    properties.Set( "posc",            posc                   );
    properties.Set( "coerce",          coerce                 );
    properties.Set( "makeDir",         makedir                );
    properties.Set( "dynamicSource",   dynSrc                 );
    properties.Set( "thirdParty",      thirdParty             );
    properties.Set( "thirdPartyMode",  thirdPartyMode         );
    properties.Set( "checkSumMode",    checkSumMode           );
    properties.Set( "checkSumType",    checkSumType           );
    properties.Set( "checkSumPreset",  checkSumPreset         );
    properties.Set( "chunkSize",       chunkSize              );
    properties.Set( "parallelChunks",  parallelChunks         );
    properties.Set( "zipArchive",      zip                    );
    properties.Set( "xcp",             xcp                    );
    properties.Set( "xcpBlockSize",    blockSize              );
    properties.Set( "delegate",        delegate               );
    properties.Set( "targetIsDir",     targetIsDir            );
    properties.Set( "preserveXAttr",   preserveXAttr          );
    properties.Set( "xrate",           config.xRate           );
    properties.Set( "xrateThreshold",  config.xRateThreshold  );
    properties.Set( "rmOnBadCksum",    rmOnBadCksum           );
    properties.Set( "continue",        continue_              );
    properties.Set( "zipAppend",       zipappend              );
    properties.Set( "addcksums",       config.AddCksVal       );
    properties.Set( "doServer",        doserver               );
 
    if( zip )
      properties.Set( "zipSource",     zipFile                );
 
    if( xcp )
      properties.Set( "nbXcpSources",  nbSources              );
 
    if( !thirdPartyTokenFile.empty() )
      properties.Set( "thirdPartyTokenFile",  thirdPartyTokenFile  );
 
    XRootDStatus st = process.AddJob( properties, results );
    if( !st.IsOK() )
    {
      std::cerr << "AddJob " << source << " -> " << target << ": ";
      std::cerr << st.ToStr() << std::endl;
    }
    resultVect.push_back( results );
    sourceFile = sourceFile->Next;
  }
 
  //----------------------------------------------------------------------------
  // Configure the copy process
  //----------------------------------------------------------------------------
  PropertyList processConfig;
  processConfig.Set( "jobType", "configuration" );
  processConfig.Set( "parallel", config.Parallel );
  process.AddJob( processConfig, 0 );
 
  //----------------------------------------------------------------------------
  // Prepare and run the copy process
  //----------------------------------------------------------------------------
  XRootDStatus st = process.Prepare();
  if( !st.IsOK() )
  {
    CleanUpResults( resultVect );
    std::cerr << "Prepare: " << st.ToStr() << std::endl;
    return st.GetShellCode();
  }
 
  st = process.Run( &progress );
  if( !st.IsOK() )
  {
    if( resultVect.size() == 1 )
      std::cerr << "Run: " << st.ToStr() << std::endl;
    else
    {
      std::vector<XrdCl::PropertyList*>::iterator it;
      uint32_t i = 1;
      uint32_t jobsRun = 0;
      uint32_t errors  = 0;
      for( it = resultVect.begin(); it != resultVect.end(); ++it, ++i )
      {
        if( !(*it)->HasProperty( "status" ) )
          continue;
 
        XRootDStatus st = (*it)->Get<XRootDStatus>("status");
        if( !st.IsOK() )
        {
          std::cerr << "Job #" << i << ": " << st.ToStr();
          ++errors;
        }
        ++jobsRun;
      }
      std::cerr << "Jobs total: " << resultVect.size();
      std::cerr << ", run: " << jobsRun;
      std::cerr << ", errors: " << errors << std::endl;
    }
    CleanUpResults( resultVect );
    return st.GetShellCode();
  }
  CleanUpResults( resultVect );
  return 0;
}