XrdClJobManager.hh

Go to the documentation of this file.

//------------------------------------------------------------------------------
// Copyright (c) 2013 by European Organization for Nuclear Research (CERN)
// Author: Lukasz Janyst <ljanyst@cern.ch>
//------------------------------------------------------------------------------
// 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/>.
//------------------------------------------------------------------------------
 
#ifndef __XRD_CL_JOB_MANAGER_HH__
#define __XRD_CL_JOB_MANAGER_HH__
 
#include <cstdint>
#include <vector>
#include <algorithm>
#include <pthread.h>
#include "XrdCl/XrdClSyncQueue.hh"
 
namespace XrdCl
{
  //----------------------------------------------------------------------------
  //----------------------------------------------------------------------------
  class Job
  {
    public:
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      virtual ~Job() {};
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      virtual void Run( void *arg ) = 0;
  };
 
  //----------------------------------------------------------------------------
  //----------------------------------------------------------------------------
  class JobManager
  {
    public:
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      JobManager( uint32_t workers )
      {
        pRunning = false;
        pWorkers.resize( workers );
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      ~JobManager()
      {
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      bool Initialize();
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      bool Finalize();
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      bool Start();
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      bool Stop();
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      void QueueJob( Job *job, void *arg = 0 )
      {
        pJobs.Put( JobHelper( job, arg ) );
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      void RunJobs();
 
      bool IsWorker()
      {
        pthread_t thread = pthread_self();
        std::vector<pthread_t>::iterator itr =
            std::find( pWorkers.begin(), pWorkers.end(), thread );
        return itr != pWorkers.end();
      }
 
    private:
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      void StopWorkers( uint32_t n );
 
      struct JobHelper
      {
        JobHelper( Job *j = 0, void *a = 0 ): job(j), arg(a) {}
        Job  *job;
        void *arg;
      };
 
      std::vector<pthread_t> pWorkers;
      SyncQueue<JobHelper>   pJobs;
      XrdSysMutex            pMutex;
      bool                   pRunning;
  };
}
 
#endif // __XRD_CL_ANY_OBJECT_HH__