XrdClParallelOperation.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_PARALLELOPERATION_HH__
#define __XRD_CL_PARALLELOPERATION_HH__
 
#include "XrdCl/XrdClOperations.hh"
#include "XrdCl/XrdClOperationHandlers.hh"
#include "XrdCl/XrdClDefaultEnv.hh"
#include "XrdCl/XrdClPostMaster.hh"
#include "XrdCl/XrdClJobManager.hh"
 
#include <atomic>
#include <condition_variable>
#include <mutex>
 
namespace XrdCl
{
  
  //----------------------------------------------------------------------------
  // Interface for different execution policies:
  // - all      : all operations need to succeed in order for the parallel
  //              operation to be successful
  // - any      : just one of the operations needs to succeed in order for
  //              the parallel operation to be successful
  // - some     : n (user defined) operations need to succeed in order for
  //              the parallel operation to be successful
  // - at least : at least n (user defined) operations need to succeed in
  //              order for the parallel operation to be successful (the
  //              user handler will be called only when all operations are
  //              resolved)
  //
  // @param status : status returned by one of the aggregated operations
  //
  // @return       : true if the status should be passed to the user handler,
  //                 false otherwise.
  //----------------------------------------------------------------------------
  struct PolicyExecutor
  {
    virtual ~PolicyExecutor()
    {
    }
 
    virtual bool Examine( const XrdCl::XRootDStatus &status ) = 0;
 
    virtual XRootDStatus Result() = 0;
  };
 
  //----------------------------------------------------------------------------
  //----------------------------------------------------------------------------
  template<bool HasHndl>
  class ParallelOperation: public ConcreteOperation<ParallelOperation, HasHndl, Resp<void>>
  {
      template<bool> friend class ParallelOperation;
 
    public:
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      template<bool from>
      ParallelOperation( ParallelOperation<from> &&obj ) :
          ConcreteOperation<ParallelOperation, HasHndl, Resp<void>>( std::move( obj ) ),
          pipelines( std::move( obj.pipelines ) ),
          policy( std::move( obj.policy ) )
      {
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      template<class Container>
      ParallelOperation( Container &&container )
      {
        static_assert( !HasHndl, "Constructor is available only operation without handler");
 
        pipelines.reserve( container.size() );
        auto begin = std::make_move_iterator( container.begin() );
        auto end   = std::make_move_iterator( container.end() );
        std::copy( begin, end, std::back_inserter( pipelines ) );
        container.clear(); // there's junk inside so we clear it
      }
 
      ~ParallelOperation()
      {
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      std::string ToString()
      {
        std::ostringstream oss;
        oss << "Parallel(";
        for( size_t i = 0; i < pipelines.size(); i++ )
        {
          oss << pipelines[i]->ToString();
          if( i + 1  != pipelines.size() )
          {
            oss << " && ";
          }
        }
        oss << ")";
        return oss.str();
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      ParallelOperation<HasHndl> All()
      {
        policy.reset( new AllPolicy() );
        return std::move( *this );
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      ParallelOperation<HasHndl> Any()
      {
        policy.reset( new AnyPolicy( pipelines.size() ) );
        return std::move( *this );
      }
 
      //------------------------------------------------------------------------
      // Set policy to `Some`
      //------------------------------------------------------------------------
      ParallelOperation<HasHndl> Some( size_t threshold )
      {
        policy.reset( new SomePolicy( pipelines.size(), threshold ) );
        return std::move( *this );
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      ParallelOperation<HasHndl> AtLeast( size_t threshold )
      {
        policy.reset( new AtLeastPolicy( pipelines.size(), threshold ) );
        return std::move( *this );
      }
 
    private:
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      struct AllPolicy : public PolicyExecutor
      {
        bool Examine( const XrdCl::XRootDStatus &status )
        {
          // keep the status in case this is the final result
          res = status;
          if( status.IsOK() ) return false;
          // we require all request to succeed
          return true;
        }
 
        XRootDStatus Result()
        {
          return res;
        }
 
        XRootDStatus res;
      };
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      struct AnyPolicy : public PolicyExecutor
      {
        AnyPolicy( size_t size) : cnt( size )
        {
        }
 
        bool Examine( const XrdCl::XRootDStatus &status )
        {
          // keep the status in case this is the final result
          res = status;
          // decrement the counter
          size_t nb = cnt.fetch_sub( 1, std::memory_order_relaxed );
          // we require just one operation to be successful
          if( status.IsOK() ) return true;
          // lets see if this is the last one?
          if( nb == 1 ) return true;
          // we still have a chance there will be one that is successful
          return false;
        }
 
        XRootDStatus Result()
        {
          return res;
        }
 
        private:
          std::atomic<size_t> cnt;
          XRootDStatus        res;
      };
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      struct SomePolicy : PolicyExecutor
      {
        SomePolicy( size_t size, size_t threshold ) : failed( 0 ), succeeded( 0 ),
                                                      threshold( threshold ), size( size )
        {
        }
 
        bool Examine( const XrdCl::XRootDStatus &status )
        {
          // keep the status in case this is the final result
          res = status;
          if( status.IsOK() )
          {
            size_t s = succeeded.fetch_add( 1, std::memory_order_relaxed );
            if( s + 1 == threshold ) return true; // we reached the threshold
            // we are not yet there
            return false;
          }
          size_t f = failed.fetch_add( 1, std::memory_order_relaxed );
          // did we drop below the threshold
          if( f == size - threshold ) return true;
          // we still have a chance there will be enough of successful operations
          return false;
        }
 
        XRootDStatus Result()
        {
          return res;
        }
 
        private:
          std::atomic<size_t> failed;
          std::atomic<size_t> succeeded;
          const size_t        threshold;
          const size_t        size;
          XRootDStatus        res;
      };
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      struct AtLeastPolicy : PolicyExecutor
      {
        AtLeastPolicy( size_t size, size_t threshold ) : pending_cnt( size ),
                                                         failed_cnt( 0 ),
                                                         failed_threshold( size - threshold )
        {
        }
 
        //----------------------------------------------------------------------
        //----------------------------------------------------------------------
        bool Examine( const XrdCl::XRootDStatus &status )
        {
          if (!status.IsOK()) {
            if (failed_cnt.fetch_add(1, std::memory_order_relaxed) == failed_threshold) {
              res = status;
              return true;
            }
          }
 
          return pending_cnt.fetch_sub(1, std::memory_order_relaxed) == 1;
        }
 
        XRootDStatus Result()
        {
          return res;
        }
 
        private:
          std::atomic<size_t> pending_cnt;
          std::atomic<size_t> failed_cnt;
          const size_t        failed_threshold;
          XRootDStatus        res;
      };
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      struct barrier_t
      {
        barrier_t() : on( true ) { }
 
        void wait()
        {
          std::unique_lock<std::mutex> lck( mtx );
          if( on ) cv.wait( lck );
        }
 
        void lift()
        {
          std::unique_lock<std::mutex> lck( mtx );
          on = false;
          cv.notify_all();
        }
 
        private:
          std::condition_variable cv;
          std::mutex              mtx;
          bool                    on;
      };
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      struct Ctx
      {
        //----------------------------------------------------------------------
        //----------------------------------------------------------------------
        Ctx( PipelineHandler *handler, PolicyExecutor  *policy  ): handler( handler ),
                                                                   policy( policy )
        {
        }
 
        //----------------------------------------------------------------------
        //----------------------------------------------------------------------
        ~Ctx()
        {
          Handle( XRootDStatus() );
        }
 
        //----------------------------------------------------------------------
        //----------------------------------------------------------------------
        inline void Examine( const XRootDStatus &st )
        {
          if( policy->Examine( st ) )
            Handle( policy->Result() );
        }
 
        //----------------------------------------------------------------------
        //---------------------------------------------------------------------
        inline void Handle( const XRootDStatus &st )
        {
            PipelineHandler* hdlr = handler.exchange( nullptr, std::memory_order_relaxed );
            if( hdlr )
            {
              barrier.wait();
              hdlr->HandleResponse( new XRootDStatus( st ), nullptr );
            }
        }
 
        //----------------------------------------------------------------------
        //----------------------------------------------------------------------
        std::atomic<PipelineHandler*> handler;
 
        //----------------------------------------------------------------------
        //----------------------------------------------------------------------
        std::unique_ptr<PolicyExecutor> policy;
 
        //----------------------------------------------------------------------
        //----------------------------------------------------------------------
        barrier_t barrier;
      };
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      struct PipelineEnd : public Job
      {
        //----------------------------------------------------------------------
        // Constructor
        //----------------------------------------------------------------------
        PipelineEnd( std::shared_ptr<Ctx>      &ctx,
                     const XrdCl::XRootDStatus &st ) : ctx( ctx ), st( st )
        {
        }
 
        //----------------------------------------------------------------------
        // Run Ctx::Examine in the thread-pool
        //----------------------------------------------------------------------
        void Run( void* )
        {
          ctx->Examine( st );
          delete this;
        }
 
        private:
          std::shared_ptr<Ctx> ctx; //< ParallelOperaion context
          XrdCl::XRootDStatus  st;  //< final status of the ParallelOperation
      };
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      inline static
      void Schedule( std::shared_ptr<Ctx> &ctx, const XrdCl::XRootDStatus &st)
      {
        XrdCl::JobManager *mgr = XrdCl::DefaultEnv::GetPostMaster()->GetJobManager();
        PipelineEnd *end = new PipelineEnd( ctx, st );
        mgr->QueueJob( end, nullptr );
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      XRootDStatus RunImpl( PipelineHandler *handler, uint16_t pipelineTimeout )
      {
        // make sure we have a valid policy for the parallel operation
        if( !policy ) policy.reset( new AllPolicy() );
 
        std::shared_ptr<Ctx> ctx =
            std::make_shared<Ctx>( handler, policy.release() );
 
        uint16_t timeout = pipelineTimeout < this->timeout ?
                           pipelineTimeout : this->timeout;
 
        for( size_t i = 0; i < pipelines.size(); ++i )
        {
          if( !pipelines[i] ) continue;
          pipelines[i].Run( timeout,
              [ctx]( const XRootDStatus &st ) mutable { Schedule( ctx, st ); } );
        }
 
        ctx->barrier.lift();
        return XRootDStatus();
      }
 
      std::vector<Pipeline>           pipelines;
      std::unique_ptr<PolicyExecutor> policy;
  };
 
  //----------------------------------------------------------------------------
  //----------------------------------------------------------------------------
  template<class Container>
  inline ParallelOperation<false> Parallel( Container &&container )
  {
    return ParallelOperation<false>( container );
  }
 
  //----------------------------------------------------------------------------
  //----------------------------------------------------------------------------
  inline void PipesToVec( std::vector<Pipeline>& )
  {
    // base case
  }
 
  //----------------------------------------------------------------------------
  // Declare PipesToVec (we need to do declare those functions ahead of
  // definitions, as they may call each other.
  //----------------------------------------------------------------------------
  template<typename ... Others>
  inline void PipesToVec( std::vector<Pipeline> &v, Operation<false> &operation,
      Others&... others );
 
  template<typename ... Others>
  inline void PipesToVec( std::vector<Pipeline> &v, Operation<true> &operation,
      Others&... others );
 
  template<typename ... Others>
  inline void PipesToVec( std::vector<Pipeline> &v, Pipeline &pipeline,
      Others&... others );
 
  //----------------------------------------------------------------------------
  // Define PipesToVec
  //----------------------------------------------------------------------------
  template<typename ... Others>
  void PipesToVec( std::vector<Pipeline> &v, Operation<false> &operation,
      Others&... others )
  {
    v.emplace_back( operation );
    PipesToVec( v, others... );
  }
 
  template<typename ... Others>
  void PipesToVec( std::vector<Pipeline> &v, Operation<true> &operation,
      Others&... others )
  {
    v.emplace_back( operation );
    PipesToVec( v, others... );
  }
 
  template<typename ... Others>
  void PipesToVec( std::vector<Pipeline> &v, Pipeline &pipeline,
      Others&... others )
  {
    v.emplace_back( std::move( pipeline ) );
    PipesToVec( v, others... );
  }
 
  //----------------------------------------------------------------------------
  //----------------------------------------------------------------------------
  template<typename ... Operations>
  inline ParallelOperation<false> Parallel( Operations&& ... operations )
  {
    constexpr size_t size = sizeof...( operations );
    std::vector<Pipeline> v;
    v.reserve( size );
    PipesToVec( v, operations... );
    return Parallel( v );
  }
}
 
#endif // __XRD_CL_OPERATIONS_HH__