XrdClXCpSrc.hh

Go to the documentation of this file.

//------------------------------------------------------------------------------
// Copyright (c) 2011-2017 by European Organization for Nuclear Research (CERN)
// Author: Michal Simon <michal.simon@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.
//------------------------------------------------------------------------------
 
#ifndef SRC_XRDCL_XRDCLXCPSRC_HH_
#define SRC_XRDCL_XRDCLXCPSRC_HH_
 
#include "XrdCl/XrdClFile.hh"
#include "XrdCl/XrdClSyncQueue.hh"
#include "XrdSys/XrdSysPthread.hh"
 
#include <atomic>
 
namespace XrdCl
{
 
class XCpCtx;
 
class XCpSrc
{
    friend class ChunkHandler;
 
  public:

    XCpSrc( uint32_t chunkSize, uint8_t parallel, int64_t fileSize, XCpCtx *ctx );

    int Start();

    void Stop()
    {
      pRunning = false;
    }

    void Delete()
    {
      XrdSysMutexHelper lck( pMtx );
      --pRefCount;
      if( !pRefCount )
      {
        lck.UnLock();
        delete this;
      }
    }

    XCpSrc* Self()
    {
      XrdSysMutexHelper lck( pMtx );
      // if Ctx is trying to increase our ref count it is possible
      // we are already in our destrutor, waiting for RemoveSrc().
      if( !pRefCount ) return nullptr;
      ++pRefCount;
      return this;
    }

    bool IsRunning()
    {
      return pRunning;
    }

    bool HasData()
    {
      XrdSysMutexHelper lck( pMtx );
      return pCurrentOffset < pBlkEnd || !pRecovered.empty() || !pOngoing.empty();
    }
 
 

    uint64_t TransferRate();

    static void DeleteChunk( PageInfo *&chunk )
    {
      if( chunk )
      {
        delete[] static_cast<char*>( chunk->GetBuffer() );
        delete   chunk;
        chunk = 0;
      }
    }
 
  private:

    virtual ~XCpSrc();

    static void* Run( void* arg );

    void StartDownloading();

    XRootDStatus Initialize();

    XRootDStatus Recover();

    XRootDStatus ReadChunks();

    void Steal( XCpSrc *src );

    XRootDStatus GetWork();

    void ReportResponse( XRootDStatus *status, PageInfo *chunk, File *handle );

    template<typename T>
    static void DeletePtr( T *&obj )
    {
      delete obj;
      obj = 0;
    }

    static bool FilesEqual( File *f1, File *f2 )
    {
      if( !f1 || !f2 ) return false;
 
      const std::string lastURL = "LastURL";
      std::string url1, url2;
 
      f1->GetProperty( lastURL, url1 );
      f2->GetProperty( lastURL, url2 );
 
      // remove cgi information
      size_t pos = url1.find( '?' );
      if( pos != std::string::npos )
        url1 = url1.substr( 0 , pos );
      pos = url2.find( '?' );
      if( pos != std::string::npos )
        url2 = url2.substr( 0 , pos );
 
      return url1 == url2;
    }

    uint32_t                      pChunkSize;

    uint8_t                       pParallel;

    int64_t                       pFileSize;

    pthread_t                     pThread;

    XCpCtx                       *pCtx;

    std::string                   pUrl;

    File                         *pFile;
 
    std::map<File*, uint8_t>      pFailed;

    uint64_t                      pCurrentOffset;

    uint64_t                      pBlkEnd;

    std::atomic<uint64_t>         pDataTransfered;

    std::map<uint64_t, uint64_t>  pOngoing;

    std::map<uint64_t, uint64_t>  pRecovered;

    SyncQueue<XRootDStatus*>      pReports;

    XrdSysRecMutex                pMtx;

    size_t                        pRefCount;

    std::atomic<bool>             pRunning;

    time_t                        pStartTime;

    time_t                        pTransferTime;

    bool                          pUsePgRead;
};
 
} /* namespace XrdCl */
 
#endif /* SRC_XRDCL_XRDCLXCPSRC_HH_ */