XrdClOperations.hh

Go to the documentation of this file.

//------------------------------------------------------------------------------
// Copyright (c) 2011-2017 by European Organization for Nuclear Research (CERN)
// Author: Krzysztof Jamrog <krzysztof.piotr.jamrog@cern.ch>,
//         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 __XRD_CL_OPERATIONS_HH__
#define __XRD_CL_OPERATIONS_HH__
 
#include <memory>
#include <stdexcept>
#include <sstream>
#include <tuple>
#include <future>
#include "XrdCl/XrdClXRootDResponses.hh"
#include "XrdCl/XrdClOperationHandlers.hh"
#include "XrdCl/XrdClArg.hh"
#include "XrdCl/XrdClOperationTimeout.hh"
#include "XrdCl/XrdClFinalOperation.hh"
#include "XrdSys/XrdSysPthread.hh"
 
#include "XrdCl/XrdClResponseJob.hh"
#include "XrdCl/XrdClJobManager.hh"
#include "XrdCl/XrdClPostMaster.hh"
#include "XrdCl/XrdClDefaultEnv.hh"
 
namespace XrdCl
{
 
  class Pipeline;
  class PipelineHandler;
 
  //----------------------------------------------------------------------------
  //----------------------------------------------------------------------------
  template<bool HasHndl>
  class Operation
  {
      // Declare friendship between templates
      template<bool>
      friend class Operation;
 
      friend std::future<XRootDStatus> Async( Pipeline, uint16_t );
 
      friend class Pipeline;
      friend class PipelineHandler;
 
    public:
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      Operation() : valid( true )
      {
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      template<bool from>
      Operation( Operation<from> && op ) :
          handler( std::move( op.handler ) ), valid( true )
      {
        if( !op.valid ) throw std::invalid_argument( "Cannot construct "
            "Operation from an invalid Operation!" );
        op.valid = false;
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      virtual ~Operation()
      {
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      virtual std::string ToString() = 0;
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      virtual Operation<HasHndl>* Move() = 0;
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      virtual Operation<true>* ToHandled() = 0;
 
    protected:
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      void Run( Timeout                                   timeout,
                std::promise<XRootDStatus>                prms,
                std::function<void(const XRootDStatus&)>  final );
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      virtual XRootDStatus RunImpl( PipelineHandler *handler, uint16_t timeout ) = 0;
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      void AddOperation( Operation<true> *op );
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      std::unique_ptr<PipelineHandler> handler;
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      bool valid;
  };
 
  //----------------------------------------------------------------------------
  //----------------------------------------------------------------------------
  typedef std::function<Operation<true>*(const XRootDStatus&)>  rcvry_func;
 
  //----------------------------------------------------------------------------
  //----------------------------------------------------------------------------
  class PipelineHandler: public ResponseHandler
  {
      template<bool> friend class Operation;
 
    public:
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      PipelineHandler( ResponseHandler   *handler );
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      PipelineHandler()
      {
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      void HandleResponseWithHosts( XRootDStatus *status, AnyObject *response,
          HostList *hostList );
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      void HandleResponse( XRootDStatus *status, AnyObject *response );
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      ~PipelineHandler()
      {
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      void AddOperation( Operation<true> *operation );
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      void Assign( const Timeout                            &timeout,
                   std::promise<XRootDStatus>                prms,
                   std::function<void(const XRootDStatus&)>  final,
                   Operation<true>                          *opr );
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      void Assign( std::function<void(const XRootDStatus&)>  final );
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      void PreparePipelineStart();
 
    private:
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      void HandleResponseImpl( XRootDStatus *status, AnyObject *response, HostList *hostList );
 
      inline void dealloc( XRootDStatus *status, AnyObject *response,
          HostList *hostList )
      {
        delete status;
        delete response;
        delete hostList;
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      std::unique_ptr<ResponseHandler> responseHandler;
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      std::unique_ptr<Operation<true>> currentOperation;
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      std::unique_ptr<Operation<true>> nextOperation;
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      Timeout timeout;
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      std::promise<XRootDStatus> prms;
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      std::function<void(const XRootDStatus&)> final;
  };
 
  //----------------------------------------------------------------------------
  //----------------------------------------------------------------------------
  class Pipeline
  {
      template<bool> friend class ParallelOperation;
      friend std::future<XRootDStatus> Async( Pipeline, uint16_t );
      friend class PipelineHandler;
 
    public:
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      Pipeline()
      {
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      Pipeline( Operation<true> *op ) :
          operation( op->Move() )
      {
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      Pipeline( Operation<true> &op ) :
          operation( op.Move() )
      {
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      Pipeline( Operation<true> &&op ) :
          operation( op.Move() )
      {
      }
 
      Pipeline( Operation<false> *op ) :
          operation( op->ToHandled() )
      {
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      Pipeline( Operation<false> &op ) :
          operation( op.ToHandled() )
      {
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      Pipeline( Operation<false> &&op ) :
          operation( op.ToHandled() )
      {
      }
 
      Pipeline( Pipeline &&pipe ) :
          operation( std::move( pipe.operation ) )
      {
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      Pipeline& operator=( Pipeline &&pipe )
      {
        operation = std::move( pipe.operation );
        return *this;
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      Pipeline& operator|=( Operation<true>&& op )
      {
        operation->AddOperation( op.Move() );
        return *this;
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      Pipeline& operator|=( Operation<false>&& op )
      {
        operation->AddOperation( op.ToHandled() );
        return *this;
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      operator Operation<true>&()
      {
        if( !bool( operation ) ) throw std::logic_error( "Invalid pipeline." );
        return *operation.get();
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      operator bool()
      {
        return bool( operation );
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      static void Stop( const XRootDStatus &status = XrdCl::XRootDStatus() );
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      static void Repeat();
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      static void Replace( Operation<false> &&opr );
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      static void Replace( Pipeline p );
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      static void Ignore();
 
    private:
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      Operation<true>* operator->()
      {
        return operation.get();
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      void Run( Timeout timeout, std::function<void(const XRootDStatus&)> final = nullptr )
      {
        if( ftr.valid() )
          throw std::logic_error( "Pipeline is already running!" );
 
        // a promise that the pipe will have a result
        std::promise<XRootDStatus> prms;
        ftr = prms.get_future();
 
        if( !operation ) std::logic_error( "Empty pipeline!" );
 
        Operation<true> *opr = operation.release();
        PipelineHandler *h = opr->handler.get();
        if( h )
          h->PreparePipelineStart();
 
        opr->Run( timeout, std::move( prms ), std::move( final ) );
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      std::unique_ptr<Operation<true>> operation;
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      std::future<XRootDStatus> ftr;
 
  };
 
  //----------------------------------------------------------------------------
  //----------------------------------------------------------------------------
  inline std::future<XRootDStatus> Async( Pipeline pipeline, uint16_t timeout = 0 )
  {
    pipeline.Run( timeout );
    return std::move( pipeline.ftr );
  }
 
  //----------------------------------------------------------------------------
  //----------------------------------------------------------------------------
  inline XRootDStatus WaitFor( Pipeline pipeline, uint16_t timeout = 0 )
  {
    return Async( std::move( pipeline ), timeout ).get();
  }
 
  //----------------------------------------------------------------------------
  //----------------------------------------------------------------------------
  template<template<bool> class Derived, bool HasHndl, typename HdlrFactory, typename ... Args>
  class ConcreteOperation: public Operation<HasHndl>
  {
      template<template<bool> class, bool, typename, typename ...>
      friend class ConcreteOperation;
 
    public:
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      ConcreteOperation( Args&&... args ) : args( std::tuple<Args...>( std::move( args )... ) ),
                                            timeout( 0 )
      {
        static_assert( !HasHndl, "It is only possible to construct operation without handler" );
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      template<bool from>
      ConcreteOperation( ConcreteOperation<Derived, from, HdlrFactory, Args...> && op ) :
        Operation<HasHndl>( std::move( op ) ), args( std::move( op.args ) ), timeout( 0 )
      {
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      template<typename Hdlr>
      Derived<true> operator>>( Hdlr &&hdlr )
      {
        return this->StreamImpl( HdlrFactory::Create( hdlr ) );
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      Derived<true> operator|( Operation<true> &op )
      {
        return PipeImpl( *this, op );
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      Derived<true> operator|( Operation<true> &&op )
      {
        return PipeImpl( *this, op );
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      Derived<true> operator|( Operation<false> &op )
      {
        return PipeImpl( *this, op );
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      Derived<true> operator|( Operation<false> &&op )
      {
        return PipeImpl( *this, op );
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      Derived<true> operator|( FinalOperation &&fo )
      {
        AllocHandler( *this );
        this->handler->Assign( fo.final );
        return this->template Transform<true>();
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      inline Operation<HasHndl>* Move()
      {
        Derived<HasHndl> *me = static_cast<Derived<HasHndl>*>( this );
        return new Derived<HasHndl>( std::move( *me ) );
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      inline Operation<true>* ToHandled()
      {
        this->handler.reset( new PipelineHandler() );
        Derived<HasHndl> *me = static_cast<Derived<HasHndl>*>( this );
        return new Derived<true>( std::move( *me ) );
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      Derived<HasHndl> Timeout( uint16_t timeout )
      {
        this->timeout = timeout;
        Derived<HasHndl> *me = static_cast<Derived<HasHndl>*>( this );
        return std::move( *me );
      }
 
    protected:
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      template<bool to>
      inline Derived<to> Transform()
      {
        Derived<HasHndl> *me = static_cast<Derived<HasHndl>*>( this );
        return Derived<to>( std::move( *me ) );
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      inline Derived<true> StreamImpl( ResponseHandler *handler )
      {
        static_assert( !HasHndl, "Operator >> is available only for operation without handler" );
        this->handler.reset( new PipelineHandler( handler ) );
        return Transform<true>();
      }
 
      //------------------------------------------------------------------------
      // Allocate handler if necessary
      //------------------------------------------------------------------------
      inline static
      void AllocHandler( ConcreteOperation<Derived, true, HdlrFactory, Args...> &me )
      {
        // nothing to do
      }
 
      //------------------------------------------------------------------------
      // Allocate handler if necessary
      //------------------------------------------------------------------------
      inline static
      void AllocHandler( ConcreteOperation<Derived, false, HdlrFactory, Args...> &me )
      {
        me.handler.reset( new PipelineHandler() );
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      inline static
      Derived<true> PipeImpl( ConcreteOperation<Derived, HasHndl, HdlrFactory,
          Args...> &me, Operation<true> &op )
      {
        AllocHandler( me ); // if HasHndl is false allocate handler
        me.AddOperation( op.Move() );
        return me.template Transform<true>();
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      inline static
      Derived<true> PipeImpl( ConcreteOperation<Derived, HasHndl, HdlrFactory,
          Args...> &me, Operation<false> &op )
      {
        AllocHandler( me ); // if HasHndl is false allocate handler
        me.AddOperation( op.ToHandled() );
        return me.template Transform<true>();
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      std::tuple<Args...> args;
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      uint16_t timeout;
    };
 
  // Out-of-line methods for class Operation
 
  template <bool HasHndl>
  void Operation<HasHndl>::Run(Timeout timeout, std::promise<XRootDStatus> prms,
                               std::function<void(const XRootDStatus &)> f)
  {
    static_assert(HasHndl, "Only an operation that has a handler can be assigned to workflow");
 
    XRootDStatus st;
    handler->Assign(timeout, std::move(prms), std::move(f), this);
    PipelineHandler *h = handler.release();
 
    try {
      st = RunImpl(h, timeout);
    } catch (const operation_expired &ex) {
      st = XRootDStatus(stError, errOperationExpired);
    } catch (const PipelineException &ex) { // probably not needed
      st = ex.GetError();
    } catch (const std::exception &ex) {
      st = XRootDStatus(stError, errInternal, 0, ex.what());
    }
 
    if (!st.IsOK()) {
      ResponseJob *job = new ResponseJob(h, new XRootDStatus(st), 0, nullptr);
      DefaultEnv::GetPostMaster()->GetJobManager()->QueueJob(job);
    }
  }
 
  template <bool HasHndl>
  void Operation<HasHndl>::AddOperation(Operation<true> *op)
  {
    if (handler)
      handler->AddOperation(op);
  }
}
 
#endif // __XRD_CL_OPERATIONS_HH__