Loading...
Searching...
No Matches
XrdClJobManager.cc
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/>.
//------------------------------------------------------------------------------
#include "XrdCl/XrdClJobManager.hh"
#include "XrdCl/XrdClLog.hh"
#include "XrdCl/XrdClDefaultEnv.hh"
#include "XrdCl/XrdClConstants.hh"
#include "XrdSys/XrdSysE2T.hh"
//------------------------------------------------------------------------------
// The thread
//------------------------------------------------------------------------------
extern "C"
{
static void *RunRunnerThread( void *arg )
{
using namespace XrdCl;
JobManager *mgr = (JobManager*)arg;
mgr->RunJobs();
return 0;
}
}
namespace XrdCl
{
//----------------------------------------------------------------------------
// Initialize the job manager
//----------------------------------------------------------------------------
bool JobManager::Initialize()
{
return true;
}
//----------------------------------------------------------------------------
// Finalize the job manager, clear the queues
//----------------------------------------------------------------------------
bool JobManager::Finalize()
{
pJobs.Clear();
return true;
}
//----------------------------------------------------------------------------
// Start the workers
//----------------------------------------------------------------------------
bool JobManager::Start()
{
XrdSysMutexHelper scopedLock( pMutex );
Log *log = DefaultEnv::GetLog();
log->Debug( JobMgrMsg, "Starting the job manager..." );
if( pRunning )
{
log->Error( JobMgrMsg, "The job manager is already running" );
return false;
}
for( uint32_t i = 0; i < pWorkers.size(); ++i )
{
int ret = ::pthread_create( &pWorkers[i], 0, ::RunRunnerThread, this );
if( ret != 0 )
{
log->Error( JobMgrMsg, "Unable to spawn a job worker thread: %s",
XrdSysE2T( errno ) );
if( i > 0 )
StopWorkers( i );
return false;
}
}
pRunning = true;
log->Debug( JobMgrMsg, "Job manager started, %zu workers", pWorkers.size() );
return true;
}
//----------------------------------------------------------------------------
// Stop the workers
//----------------------------------------------------------------------------
bool JobManager::Stop()
{
XrdSysMutexHelper scopedLock( pMutex );
Log *log = DefaultEnv::GetLog();
log->Debug( JobMgrMsg, "Stopping the job manager..." );
if( !pRunning )
{
log->Error( JobMgrMsg, "The job manager is not running" );
return false;
}
StopWorkers( pWorkers.size() );
pRunning = false;
log->Debug( JobMgrMsg, "Job manager stopped" );
return true;
}
//----------------------------------------------------------------------------
// Stop all workers up to n'th
//----------------------------------------------------------------------------
void JobManager::StopWorkers( uint32_t n )
{
Log *log = DefaultEnv::GetLog();
for( uint32_t i = 0; i < n; ++i )
{
void *threadRet;
log->Dump( JobMgrMsg, "Stopping worker #%d...", i );
int rc = pthread_cancel( pWorkers[i] );
if( rc != 0 )
{
log->Error( TaskMgrMsg, "Unable to cancel worker #%d: %s", i,
XrdSysE2T( errno ) );
if( rc == ESRCH ) continue;
abort();
}
rc = pthread_join( pWorkers[i], (void**)&threadRet );
if( rc != 0 )
{
log->Error( TaskMgrMsg, "Unable to join worker #%d: %s", i,
XrdSysE2T( errno ) );
if( rc == ESRCH ) continue;
abort();
}
log->Dump( JobMgrMsg, "Worker #%d stopped", i );
}
}
//----------------------------------------------------------------------------
// Initialize the job manager
//----------------------------------------------------------------------------
void JobManager::RunJobs()
{
pthread_setcanceltype( PTHREAD_CANCEL_DEFERRED, 0 );
for( ;; )
{
JobHelper h = pJobs.Get();
pthread_setcancelstate( PTHREAD_CANCEL_DISABLE, 0 );
h.job->Run( h.arg );
pthread_setcancelstate( PTHREAD_CANCEL_ENABLE, 0 );
}
}
}
XrdClConstants.hh
XrdClDefaultEnv.hh
RunRunnerThread
static void * RunRunnerThread(void *arg)
Definition XrdClJobManager.cc:30
XrdClJobManager.hh
XrdClLog.hh
XrdSysE2T
const char * XrdSysE2T(int errcode)
Definition XrdSysE2T.cc:104
XrdSysE2T.hh
XrdCl::DefaultEnv::GetLog
static Log * GetLog()
Get default log.
Definition XrdClDefaultEnv.cc:493
XrdCl::JobManager
A synchronized queue.
Definition XrdClJobManager.hh:51
XrdCl::JobManager::Finalize
bool Finalize()
Finalize the job manager, clear the queues.
Definition XrdClJobManager.cc:52
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::JobManager::RunJobs
void RunJobs()
Run the jobs.
Definition XrdClJobManager.cc:146
XrdCl::JobManager::Stop
bool Stop()
Stop the workers.
Definition XrdClJobManager.cc:93
XrdCl::Job::Run
virtual void Run(void *arg)=0
The job logic.
XrdCl::Log
Handle diagnostics.
Definition XrdClLog.hh:101
XrdCl::Log::Error
void Error(uint64_t topic, const char *format,...)
Report an error.
Definition XrdClLog.cc:231
XrdCl::Log::Dump
void Dump(uint64_t topic, const char *format,...)
Print a dump message.
Definition XrdClLog.cc:299
XrdCl::Log::Debug
void Debug(uint64_t topic, const char *format,...)
Print a debug message.
Definition XrdClLog.cc:282
XrdSysMutexHelper
Definition XrdSysPthread.hh:263
XrdCl
Definition XrdClAction.hh:34
XrdCl::TaskMgrMsg
const uint64_t TaskMgrMsg
Definition XrdClConstants.hh:38
XrdCl::JobMgrMsg
const uint64_t JobMgrMsg
Definition XrdClConstants.hh:42