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 "XrdClArg.hh"
#include "XrdSys/XrdSysPthread.hh"
 
namespace XrdCl
{
 
  template<bool HasHndl> class Operation;
 
  class Pipeline;
 
  //----------------------------------------------------------------------------
  //----------------------------------------------------------------------------
  class PipelineHandler: public ResponseHandler
  {
      template<bool> friend class Operation;
 
    public:
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      PipelineHandler( ResponseHandler *handler, bool own );
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      PipelineHandler();
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      void HandleResponseWithHosts( XRootDStatus *status, AnyObject *response,
          HostList *hostList );
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      void HandleResponse( XRootDStatus *status, AnyObject *response );
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      ~PipelineHandler();
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      void AddOperation( Operation<true> *operation );
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      void Assign( std::promise<XRootDStatus>               prms,
                   std::function<void(const XRootDStatus&)> final );
 
    private:
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      void HandleResponseImpl( XRootDStatus *status, AnyObject *response,
          HostList *hostList = nullptr );
 
      inline void dealloc( XRootDStatus *status, AnyObject *response,
          HostList *hostList )
      {
        delete status;
        delete response;
        delete hostList;
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      ResponseHandler *responseHandler;
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      bool ownHandler;
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      std::unique_ptr<Operation<true>> nextOperation;
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      std::promise<XRootDStatus> prms;
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      std::function<void(const XRootDStatus&)> final;
  };
 
  //----------------------------------------------------------------------------
  //----------------------------------------------------------------------------
  template<bool HasHndl>
  class Operation
  {
      // Declare friendship between templates
      template<bool>
      friend class Operation;
 
      friend std::future<XRootDStatus> Async( Pipeline );
 
      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( std::promise<XRootDStatus>                prms,
                std::function<void(const XRootDStatus&)>  final )
      {
        static_assert(HasHndl, "Only an operation that has a handler can be assigned to workflow");
        handler->Assign( std::move( prms ), std::move( final ) );
        XRootDStatus st = RunImpl();
        if( st.IsOK() ) handler.release();
        else
          ForceHandler( st );
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      virtual XRootDStatus RunImpl() = 0;
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      void ForceHandler( const XRootDStatus &status )
      {
        handler->HandleResponse( new XRootDStatus( status ), nullptr );
        // HandleResponse already freed the memory so we have to
        // release the unique pointer
        handler.release();
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      void AddOperation( Operation<true> *op )
      {
        if( handler )
        {
          handler->AddOperation( op );
        }
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      std::unique_ptr<PipelineHandler> handler;
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      bool valid;
  };
 
  //----------------------------------------------------------------------------
  //----------------------------------------------------------------------------
  class Pipeline
  {
      template<bool> friend class ParallelOperation;
      friend std::future<XRootDStatus> Async( Pipeline );
 
    public:
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      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;
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      operator Operation<true>&()
      {
        if( !bool( operation ) ) throw std::logic_error( "Invalid pipeline." );
        return *operation.get();
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      operator bool()
      {
        return bool( operation );
      }
 
    private:
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      Operation<true>* operator->()
      {
        return operation.get();
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      void Run( 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();
        operation->Run( std::move( prms ), std::move( final ) );
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      std::unique_ptr<Operation<true>> operation;
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      std::future<XRootDStatus> ftr;
 
  };
 
  //----------------------------------------------------------------------------
  //----------------------------------------------------------------------------
  inline std::future<XRootDStatus> Async( Pipeline pipeline )
  {
    pipeline.Run();
    return std::move( pipeline.ftr );
  }
 
  //----------------------------------------------------------------------------
  //----------------------------------------------------------------------------
  inline XRootDStatus WaitFor( Pipeline pipeline )
  {
    return Async( std::move( pipeline ) ).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 )... ) )
      {
        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 ) )
      {
 
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      template<typename Hdlr>
      Derived<true> operator>>( Hdlr &&hdlr )
      {
        // check if the resulting handler should be owned by us or by the user,
        // if the user passed us directly a ResponseHandler it's owned by the
        // user, otherwise we need to wrap the argument in a handler and in this
        // case the resulting handler will be owned by us
        constexpr bool own = !IsResponseHandler<Hdlr>::value;
        return this->StreamImpl( HdlrFactory::Create( hdlr ), own );
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      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 );
      }
 
    protected:
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      Operation<HasHndl>* Move()
      {
        Derived<HasHndl> *me = static_cast<Derived<HasHndl>*>( this );
        return new Derived<HasHndl>( std::move( *me ) );
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      Operation<true>* ToHandled()
      {
        this->handler.reset( new PipelineHandler() );
        Derived<HasHndl> *me = static_cast<Derived<HasHndl>*>( this );
        return new Derived<true>( std::move( *me ) );
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      template<bool to>
      Derived<to> Transform()
      {
        Derived<HasHndl> *me = static_cast<Derived<HasHndl>*>( this );
        return Derived<to>( std::move( *me ) );
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      inline Derived<true> StreamImpl( ResponseHandler *handler, bool own )
      {
        static_assert( !HasHndl, "Operator >> is available only for operation without handler" );
        this->handler.reset( new PipelineHandler( handler, own ) );
        return Transform<true>();
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      inline static
      Derived<true> PipeImpl( ConcreteOperation<Derived, true, HdlrFactory,
          Args...> &me, Operation<true> &op )
      {
        me.AddOperation( op.Move() );
        return me.template Transform<true>();
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      inline static
      Derived<true> PipeImpl( ConcreteOperation<Derived, true, HdlrFactory,
          Args...> &me, Operation<false> &op )
      {
        me.AddOperation( op.ToHandled() );
        return me.template Transform<true>();
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      inline static
      Derived<true> PipeImpl( ConcreteOperation<Derived, false, HdlrFactory,
          Args...> &me, Operation<true> &op )
      {
        me.handler.reset( new PipelineHandler() );
        me.AddOperation( op.Move() );
        return me.template Transform<true>();
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      inline static
      Derived<true> PipeImpl( ConcreteOperation<Derived, false, HdlrFactory,
          Args...> &me, Operation<false> &op )
      {
        me.handler.reset( new PipelineHandler() );
        me.AddOperation( op.ToHandled() );
        return me.template Transform<true>();
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      std::tuple<Args...> args;
    };
}
 
#endif // __XRD_CL_OPERATIONS_HH__