Loading...
Searching...
No Matches
XrdEcThreadPool.hh
Go to the documentation of this file.
//------------------------------------------------------------------------------
// Copyright (c) 2011-2014 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.
//------------------------------------------------------------------------------
#include "XrdCl/XrdClJobManager.hh"
#include <future>
#include <type_traits>
#ifndef SRC_XRDEC_XRDECTHREADPOOL_HH_
#define SRC_XRDEC_XRDECTHREADPOOL_HH_
namespace XrdEc
{
//---------------------------------------------------------------------------
// A theread pool class for the XrdEc module
//---------------------------------------------------------------------------
class ThreadPool
{
private:
// This is the type which holds sequences
template<int ... Is> struct sequence {};
// First define the template signature
template <int ... Ns> struct seq_gen;
// Recursion case
template <int I, int ... Ns>
struct seq_gen<I, Ns...>
{
using type = typename seq_gen<I - 1, I - 1, Ns...>::type;
};
// Recursion abort
template <int ... Ns>
struct seq_gen<0, Ns...>
{
using type = sequence<Ns...>;
};
// call functional with arguments in a tuple (implementation)
template <typename FUNC, typename TUPL, int ... INDICES>
inline static auto tuple_call_impl( FUNC &func, TUPL &args, sequence<INDICES...> ) -> decltype( func( std::move( std::get<INDICES>( args ) )... ) )
{
return func( std::move( std::get<INDICES>( args ) )... );
}
// call functional with argumetns packaged in a tuple
template <typename FUNC, typename ... ARGs>
inline static auto tuple_call( FUNC &func, std::tuple<ARGs...> &tup ) ->decltype( tuple_call_impl( func, tup, typename seq_gen<sizeof...(ARGs)>::type{} ) )
{
return tuple_call_impl( func, tup, typename seq_gen<sizeof...(ARGs)>::type{} );
}
//-----------------------------------------------------------------------
// Helper class implementing a job containing any functional and its
// arguments.
//-----------------------------------------------------------------------
template<typename FUNC, typename RET, typename ... ARGs>
class AnyJob : public XrdCl::Job
{
//---------------------------------------------------------------------
// Run the functional (returning void) with the packaged arguments
//---------------------------------------------------------------------
static inline void RunImpl( FUNC func, std::tuple<ARGs...> &args, std::promise<void> &prms )
{
tuple_call( func, args );
prms.set_value();
}
//---------------------------------------------------------------------
// Run the functional (returning anything but void) with the packaged
// arguments
//---------------------------------------------------------------------
template<typename RETURN>
static inline void RunImpl( FUNC func, std::tuple<ARGs...> &args, std::promise<RETURN> &prms )
{
prms.set_value( tuple_call( func, args ) );
}
public:
//-------------------------------------------------------------------
//-------------------------------------------------------------------
AnyJob( FUNC func, ARGs... args ) : func( std::move( func ) ),
args( std::tuple<ARGs...>( std::move( args )... ) )
{
}
//-------------------------------------------------------------------
//-------------------------------------------------------------------
void Run( void *arg )
{
RunImpl( this->func, this->args, this->prms );
delete this;
}
//-------------------------------------------------------------------
//-------------------------------------------------------------------
std::future<RET> GetFuture()
{
return prms.get_future();
}
protected:
FUNC func; //< the functional
std::tuple<ARGs...> args; //< the arguments
std::promise<RET> prms; //< the promiss that there will be a result
};
public:
//-----------------------------------------------------------------------
//-----------------------------------------------------------------------
~ThreadPool()
{
threadpool.Stop();
threadpool.Finalize();
}
//-----------------------------------------------------------------------
//-----------------------------------------------------------------------
static ThreadPool& Instance()
{
static ThreadPool instance;
return instance;
}
//-----------------------------------------------------------------------
//-----------------------------------------------------------------------
template<typename FUNC, typename ... ARGs>
inline std::future<std::invoke_result_t<FUNC, ARGs...>>
Execute( FUNC func, ARGs... args )
{
using RET = std::invoke_result_t<FUNC, ARGs...>;
auto *job = new AnyJob<FUNC, RET, ARGs...>( func, std::move( args )... );
std::future<RET> ftr = job->GetFuture();
threadpool.QueueJob( job, nullptr );
return ftr;
}
private:
//-----------------------------------------------------------------------
//-----------------------------------------------------------------------
ThreadPool() : threadpool( 64 )
{
threadpool.Initialize();
threadpool.Start();
}
XrdCl::JobManager threadpool; //< the thread-pool itself
};
}
#endif /* SRC_XRDEC_XRDECTHREADPOOL_HH_ */
XrdClJobManager.hh
I
#define I(x)
Definition XrdSectestClient.cc:65
XrdCl::JobManager
A synchronized queue.
Definition XrdClJobManager.hh:51
XrdCl::JobManager::Start
bool Start()
Start the workers.
Definition XrdClJobManager.cc:61
XrdCl::JobManager::Initialize
bool Initialize()
Initialize the job manager.
Definition XrdClJobManager.cc:44
XrdCl::Job
Interface for a job to be run by the job manager.
Definition XrdClJobManager.hh:34
XrdEc::ThreadPool
Definition XrdEcThreadPool.hh:39
XrdEc::ThreadPool::Instance
static ThreadPool & Instance()
Singleton access.
Definition XrdEcThreadPool.hh:152
XrdEc::ThreadPool::~ThreadPool
~ThreadPool()
Destructor.
Definition XrdEcThreadPool.hh:143
XrdEc::ThreadPool::Execute
std::future< std::invoke_result_t< FUNC, ARGs... > > Execute(FUNC func, ARGs... args)
Schedule a functional (together with its arguments) for execution.
Definition XrdEcThreadPool.hh:163
XrdEc
Definition XrdClZipArchive.hh:45