XrdScheduler Class Reference

#include <XrdScheduler.hh>

Inheritance diagram for XrdScheduler:

Collaboration diagram for XrdScheduler:

Public Member Functions
	XrdScheduler (int minw=3, int maxw=128, int maxi=12)
	XrdScheduler (XrdSysError *eP, XrdOucTrace *tP, int minw=8, int maxw=8192, int maxi=780)
	XrdScheduler (XrdSysError *eP, XrdSysTrace *tP, int minw=8, int maxw=8192, int maxi=780)
	~XrdScheduler ()
int	Active ()
void	Cancel (XrdJob *jp)
int	canStick ()
void	DoIt ()
pid_t	Fork (const char *id)
void *	Reaper ()
void	Run ()
void	Schedule (int num, XrdJob *jfirst, XrdJob *jlast)
void	Schedule (XrdJob *jp)
void	Schedule (XrdJob *jp, time_t atime)
void	setNproc (const bool limlower)
void	setParms (int minw, int maxw, int avlt, int maxi, int once=0)
void	Start ()
int	Stats (char *buff, int blen, int do_sync=0)
void	TimeSched ()
Public Member Functions inherited from XrdJob
	XrdJob (const char *desc="")
virtual	~XrdJob ()

Public Attributes
int	max_QLength
int	num_Jobs
int	num_Limited
int	num_TCreate
int	num_TDestroy
Public Attributes inherited from XrdJob
const char *	Comment
XrdJob *	NextJob

Detailed Description

Definition at line 46 of file XrdScheduler.hh.

Constructor & Destructor Documentation

XrdScheduler() [1/3]

XrdScheduler::XrdScheduler	(	XrdSysError *	eP,
		XrdSysTrace *	tP,
		int	minw = 8,
		int	maxw = 8192,
		int	maxi = 780 )

Definition at line 97 of file XrdScheduler.cc.

              : XrdJob("underused thread monitor"),
                 XrdTraceOld(0), WorkAvail(0, "sched work"),
                 TimerRings(TimerMutex)
{
   Boot(eP, tP, minw, maxw, maxi);
}

References XrdJob::XrdJob().

Here is the call graph for this function:

XrdScheduler() [2/3]

XrdScheduler::XrdScheduler	(	XrdSysError *	eP,
		XrdOucTrace *	tP,
		int	minw = 8,
		int	maxw = 8192,
		int	maxi = 780 )

Definition at line 109 of file XrdScheduler.cc.

              : XrdJob("underused thread monitor"),
                XrdTraceOld(tP), WorkAvail(0, "sched work"),
                TimerRings(TimerMutex)
{
 
// Invoke the main initialization function with a new style trace object
//
   Boot(eP, new XrdSysTrace("Xrd", eP->logger()), minw, maxw, maxi);
}

References XrdJob::XrdJob(), and XrdSysError::logger().

Here is the call graph for this function:

XrdScheduler() [3/3]

XrdScheduler::XrdScheduler	(	int	minw = 3,
		int	maxw = 128,
		int	maxi = 12 )

Definition at line 125 of file XrdScheduler.cc.

              : XrdJob("underused thread monitor"),
                XrdTraceOld(0), WorkAvail(0, "sched work"),
                TimerRings(TimerMutex)
{
   XrdSysLogger *Logger;
   int eFD;
 
// Get a file descriptor mirroring standard error
//
#if ( defined(__linux__) || defined(__GNU__) ) && defined(F_DUPFD_CLOEXEC)
   eFD = fcntl(STDERR_FILENO, F_DUPFD_CLOEXEC, 0);
#else
   eFD = dup(STDERR_FILENO);
   fcntl(eFD, F_SETFD, FD_CLOEXEC);
#endif
 
// Now we need to get a logger object. We make this a real dumb one.
//
   Logger = new XrdSysLogger(eFD, 0);
   XrdLog = new XrdSysError(Logger);
 
// Now get a trace object
//
   XrdTrace = new XrdSysTrace("Xrd", Logger);
   if (getenv("XRDDEBUG") != 0) XrdTrace->What = TRACE_SCHED;
 
// Set remaining values. We do no use maximum possible threads here.
//
   Init(minw, maxw, maxi);
}

References XrdJob::XrdJob(), Logger, and TRACE_SCHED.

Here is the call graph for this function:

~XrdScheduler()

XrdScheduler::~XrdScheduler

(

)

Definition at line 180 of file XrdScheduler.cc.

{
}

Member Function Documentation

Active()

int XrdScheduler::Active ( )

inline

Definition at line 50 of file XrdScheduler.hh.

{return num_Workers - idl_Workers + num_JobsinQ;}

Cancel()

void XrdScheduler::Cancel

(

XrdJob *

jp

)

Definition at line 188 of file XrdScheduler.cc.

{
   XrdJob *p, *pp = 0;
 
// Lock the queue
//
   TimerMutex.Lock();
 
// Find the matching job, if any
//
   p = TimerQueue;
   while(p && p != jp) {pp = p; p = p->NextJob;}
 
// Delete the job element
//
   if (p)
      {if (pp) pp->NextJob = p->NextJob;
          else  TimerQueue = p->NextJob;
       TRACE(SCHED, "time event " <<jp->Comment <<" cancelled");
      }
 
// All done
//
   TimerMutex.UnLock();
}

References XrdJob::XrdJob(), XrdJob::Comment, XrdJob::NextJob, and TRACE.

Referenced by Schedule(), and setParms().

Here is the call graph for this function:

Here is the caller graph for this function:

canStick()

int XrdScheduler::canStick ( )

inline

Definition at line 54 of file XrdScheduler.hh.

                         {return  num_Workers              < stk_Workers
                              || (num_Workers-idl_Workers) < stk_Workers;}

DoIt()

void XrdScheduler::DoIt ( )

virtual

Implements XrdJob.

Definition at line 218 of file XrdScheduler.cc.

{
   int num_kill, num_idle;
 
// Now check if there are too many idle threads (kill them if there are)
//
   if (!num_JobsinQ)
      {DispatchMutex.Lock(); num_idle = idl_Workers; DispatchMutex.UnLock();
       num_kill = num_idle - min_Workers;
       TRACE(SCHED, num_Workers <<" threads; " <<num_idle <<" idle");
       if (num_kill > 0)
          {if (num_kill > 1) num_kill = num_kill/2;
           SchedMutex.Lock();
           num_Layoffs = num_kill;
           while(num_kill--) WorkAvail.Post();
           SchedMutex.UnLock();
          }
      }
 
// Check if we should reschedule ourselves
//
   if (max_Workidl > 0) Schedule((XrdJob *)this, max_Workidl+time(0));
}

References XrdJob::XrdJob(), Schedule(), and TRACE.

Here is the call graph for this function:

Fork()

pid_t XrdScheduler::Fork

(

const char *

id

)

Definition at line 248 of file XrdScheduler.cc.

{
   static int  retc, ReaperStarted = 0;
   pthread_t tid;
   pid_t pid;
 
// Fork
//
   if ((pid = fork()) < 0)
      {XrdLog->Emsg("Scheduler",errno,"fork to handle",id);
       return pid;
      }
   if (!pid) return pid;
 
// Obtain the status of the reaper thread.
//
   ReaperMutex.Lock();
   firstPID = new XrdSchedulerPID(pid, firstPID);
   retc = ReaperStarted;
   ReaperStarted = 1;
   ReaperMutex.UnLock();
 
// Start the reaper thread if it has not started.
//
   if (!retc)
      if ((retc = XrdSysThread::Run(&tid, XrdStartReaper, (void *)this,
                                    0, "Process reaper")))
         {XrdLog->Emsg("Scheduler", retc, "create reaper thread");
          ReaperStarted = 0;
         }
 
   return pid;
}

References XrdSysThread::Run(), and XrdStartReaper().

Here is the call graph for this function:

Reaper()

void * XrdScheduler::Reaper

(

)

Definition at line 286 of file XrdScheduler.cc.

{
   int status;
   pid_t pid;
   XrdSchedulerPID *tp, *ptp, *xtp;
#if defined(__APPLE__) && !defined(MAC_OS_X_VERSION_10_5)
   struct timespec ts = { 1, 0 };
#else
   sigset_t Sset;
   int signum;
 
// Set up for signal handling. Note: main() must block this signal at start)
//
   sigemptyset(&Sset);
   sigaddset(&Sset, SIGCHLD);
#endif
 
// Wait for all outstanding children
//
   do {ReaperMutex.Lock();
       tp = firstPID; ptp = 0;
       while(tp)
            {do {pid = waitpid(tp->pid, &status, WNOHANG);}
                while (pid < 0 && errno == EINTR);
             if (pid > 0)
                {if (TRACING(TRACE_SCHED)) traceExit(pid, status);
                 xtp = tp; tp = tp->next;
                 if (ptp) ptp->next = tp;
                    else   firstPID = tp;
                 delete xtp;
                } else {ptp = tp; tp = tp->next;}
             }
       ReaperMutex.UnLock();
#if defined(__APPLE__) && !defined(MAC_OS_X_VERSION_10_5)
       // Mac OS X sigwait() is broken on <= 10.4.
      } while (nanosleep(&ts, 0) <= 0);
#else
      } while(sigwait(&Sset, &signum) >= 0);
#endif
   return (void *)0;
}

References XrdSchedulerPID::next, XrdSchedulerPID::pid, TRACE_SCHED, TRACING, and ts().

Referenced by XrdStartReaper().

Here is the call graph for this function:

Here is the caller graph for this function:

Run()

void XrdScheduler::Run

(

)

Definition at line 332 of file XrdScheduler.cc.

{
   int waiting;
   XrdJob *jp;
 
// Wait for work then do it (an endless task for a worker thread)
//
   do {do {DispatchMutex.Lock();          idl_Workers++;DispatchMutex.UnLock();
           WorkAvail.Wait();
           DispatchMutex.Lock();waiting = --idl_Workers;DispatchMutex.UnLock();
           SchedMutex.Lock();
           if ((jp = WorkFirst))
              {if (!(WorkFirst = jp->NextJob)) WorkLast = 0;
               if (num_JobsinQ) num_JobsinQ--;
                  else XrdLog->Emsg("Scheduler","Job queue count underflow!");
              } else {
               num_JobsinQ = 0;
               if (num_Layoffs > 0)
                  {num_Layoffs--;
                   if (waiting)
                      {num_TDestroy++; num_Workers--;
                       TRACE(SCHED, "terminating thread; workers=" <<num_Workers);
                       SchedMutex.UnLock();
                       return;
                      }
                  }
              }
           SchedMutex.UnLock();
          } while(!jp);
 
    // Check if we should hire a new worker (we always want 1 idle thread)
    // before running this job.
    //
       if (!waiting) hireWorker();
       if (TRACING(TRACE_SCHED) && *(jp->Comment) != '.')
          {TRACE(SCHED, "running " <<jp->Comment <<" inq=" <<num_JobsinQ);}
       jp->DoIt();
      } while(1);
}

References XrdJob::XrdJob(), XrdJob::Comment, XrdJob::DoIt(), XrdJob::NextJob, num_TDestroy, TRACE, TRACE_SCHED, and TRACING.

Referenced by XrdStartWorking().

Here is the call graph for this function:

Here is the caller graph for this function:

Schedule() [1/3]

void XrdScheduler::Schedule	(	int	num,
		XrdJob *	jfirst,
		XrdJob *	jlast )

Definition at line 407 of file XrdScheduler.cc.

{
 
// Lock down our data area
//
   SchedMutex.Lock();
 
// Place the request list on the queue
//
   jlast->NextJob = 0;
   if (WorkFirst)
      {WorkLast->NextJob = jfirst;
       WorkLast = jlast;
      } else {
       WorkFirst = jfirst;
       WorkLast  = jlast;
      }
 
// Calculate statistics
//
   num_Jobs    += numjobs;
   num_JobsinQ += numjobs;
   if (num_JobsinQ > max_QLength) max_QLength = num_JobsinQ;
 
// Indicate number of jobs to work on
//
   while(numjobs--) WorkAvail.Post();
 
// Unlock the data area and return
//
   SchedMutex.UnLock();
}

References XrdJob::XrdJob(), max_QLength, XrdJob::NextJob, and num_Jobs.

Here is the call graph for this function:

Schedule() [2/3]

void XrdScheduler::Schedule

(

XrdJob *

jp

)

Definition at line 376 of file XrdScheduler.cc.

{
// Lock down our data area
//
   SchedMutex.Lock();
 
// Place the request on the queue and broadcast it
//
   jp->NextJob  = 0;
   if (WorkFirst)
      {WorkLast->NextJob = jp;
       WorkLast = jp;
      } else {
       WorkFirst = jp;
       WorkLast  = jp;
      }
   WorkAvail.Post();
 
// Calculate statistics
//
   num_Jobs++;
   num_JobsinQ++;
   if (num_JobsinQ > max_QLength) max_QLength = num_JobsinQ;
 
// Unlock the data area and return
//
   SchedMutex.UnLock();
}

References XrdJob::XrdJob(), max_QLength, XrdJob::NextJob, and num_Jobs.

Referenced by XrdXrootdJob::XrdXrootdJob(), XrdXrootdPrepare::XrdXrootdPrepare(), XrdPollE::Disable(), XrdPollPoll::Disable(), XrdNetRefresh::DoIt(), DoIt(), XrdXrootdCallBack::Done(), mainAccept(), mainAdmin(), setParms(), XrdNetRefresh::Start(), XrdPollE::Start(), XrdPollPoll::Start(), Start(), and TimeSched().

Here is the call graph for this function:

Here is the caller graph for this function:

Schedule() [3/3]

void XrdScheduler::Schedule	(	XrdJob *	jp,
		time_t	atime )

Definition at line 442 of file XrdScheduler.cc.

{
   XrdJob *pp = 0, *p;
 
// Cancel this event, if scheduled
//
   Cancel(jp);
 
// Lock the queue
//
   if (TRACING(TRACE_SCHED) && *(jp->Comment) != '.')
      {TRACE(SCHED, "scheduling " <<jp->Comment <<" in " <<atime-time(0) <<" seconds");}
   jp->SchedTime = atime;
   TimerMutex.Lock();
 
// Find the insertion point for the work element
//
   p = TimerQueue;
   while(p && p->SchedTime <= atime) {pp = p; p = p->NextJob;}
 
// Insert the job element
//
   jp->NextJob = p;
   if (pp)  pp->NextJob = jp;
      else {TimerQueue = jp; TimerRings.Signal();}
 
// All done
//
   TimerMutex.UnLock();
}

References XrdJob::XrdJob(), Cancel(), XrdJob::Comment, XrdJob::NextJob, TRACE, TRACE_SCHED, and TRACING.

Here is the call graph for this function:

setNproc()

void XrdScheduler::setNproc

(

const bool

limlower

)

Definition at line 477 of file XrdScheduler.cc.

{
  // If supported change the NPROC resource limit and set max_Workers.
  // Caller can select leaving or increasing the limit, or potentially
  // setting a more restrictive limit than the current one.
 
  // If this method is called the setParms method should be called after, so
  // that a caller supplied value of max_Workers can override the value here.
 
  // We attempt to set the limit to our maximum supported threads in the
  // XrdScheduler (with a margin for pid_max).
 
// Reset the soft limit applied to our process concerning the system wide
// number threads for our uid (Linux only).
//
#if ( defined(__linux__) || defined(__GNU__) || (defined(__FreeBSD_kernel__) && defined(__GLIBC__)) ) && defined(RLIMIT_NPROC)
 
   struct rlimit rlim;
 
// First determine the absolute maximum we can have
//
   rlim_t theMax = MAX_SCHED_PROCS;
   int pdFD, rdsz;
   if ((pdFD = open("/proc/sys/kernel/pid_max", O_RDONLY)) >= 0)
      {char pmBuff[32];
       if ((rdsz = read(pdFD, pmBuff, sizeof(pmBuff))) > 0)
          {rdsz = atoi(pmBuff);
           if (rdsz < 16384) theMax = 16384; // This is unlikely
              else if (rdsz < MAX_SCHED_PROCS)
                      theMax = static_cast<rlim_t>(rdsz-2000);
          }
       close(pdFD);
      }
 
// We allow disabling the NPROC setting entirely.
//
   const bool setnp = (getenv("XRDLEAVENPROC") == 0);
 
// Get the resource thread limit and set to maximum. In Linux this may be -1
// to indicate useless infnity, so we have to come up with a number, sigh.
//
   if (setnp && !getrlimit(RLIMIT_NPROC, &rlim))
      {if (rlim.rlim_max == RLIM_INFINITY || rlim.rlim_max > theMax)
          {if (limlower || (rlim.rlim_cur != RLIM_INFINITY && rlim.rlim_cur < theMax))
              {rlim.rlim_cur = theMax;
               setrlimit(RLIMIT_NPROC, &rlim);
              }
          } else {
           if (rlim.rlim_cur != rlim.rlim_max)
              {rlim.rlim_cur = rlim.rlim_max;
               setrlimit(RLIMIT_NPROC, &rlim);
              }
          }
      }
 
// Readjust our internal maximum to be the actual maximum
//
   if (!getrlimit(RLIMIT_NPROC, &rlim))
      {if (rlim.rlim_cur == RLIM_INFINITY || rlim.rlim_cur > theMax)
               max_Workers = static_cast<int>(theMax);
          else max_Workers = static_cast<int>(rlim.rlim_cur);
      }
 
// The above may allow way too many threads. We make sure that the default
// maximum threads is adhered to.
//
   if (max_Workers > DFL_SCHED_PROCS) max_Workers = DFL_SCHED_PROCS;
#endif
}

References close, DFL_SCHED_PROCS, MAX_SCHED_PROCS, open, and read.

setParms()

void XrdScheduler::setParms	(	int	minw,
		int	maxw,
		int	avlt,
		int	maxi,
		int	once = 0 )

Definition at line 551 of file XrdScheduler.cc.

{
   static int isSet = 0;
 
// Lock the data area and check for 1-time set
//
   SchedMutex.Lock();
   if (once && isSet) {SchedMutex.UnLock(); return;}
   isSet = 1;
 
// get a consistent view of all the values
//
   if (maxw <= 0) maxw = max_Workers;
   if (minw < 0)  minw = min_Workers;
   if (minw > maxw) minw = maxw;
   if (avlw < 0) avlw = maxw/4*3;
      else if (avlw > maxw) avlw = maxw;
 
// Set the values
//
   min_Workers = minw;
   max_Workers = maxw;
   stk_Workers = maxw - avlw;
   if (maxi >=0)  max_Workidl = maxi;
 
// Unlock the data area
//
   SchedMutex.UnLock();
 
// If we have an idle interval, schedule the idle check
//
   if (maxi > 0)
      {Cancel((XrdJob *)this);
       Schedule((XrdJob *)this, (time_t)maxi+time(0));
      }
 
// Debug the info
//
   TRACE(SCHED,"Set min_Workers=" <<min_Workers <<" max_Workers=" <<max_Workers);
   TRACE(SCHED,"Set stk_Workers=" <<stk_Workers <<" max_Workidl=" <<max_Workidl);
}

References XrdJob::XrdJob(), Cancel(), Schedule(), and TRACE.

Here is the call graph for this function:

Start()

void XrdScheduler::Start

(

)

Definition at line 597 of file XrdScheduler.cc.

{
   int retc, numw;
   pthread_t tid;
 
// Provide ABI compatibility for XrdOucTrace which is deprecated!
//
   if (getenv("XRDDEBUG") != 0) XrdTrace->What = TRACE_SCHED;
      else if (XrdTraceOld) XrdTrace->What |= XrdTraceOld->What;
 
// Start a time based scheduler
//
   if ((retc = XrdSysThread::Run(&tid, XrdStartTSched, (void *)this,
                                 XRDSYSTHREAD_BIND, "Time scheduler")))
      XrdLog->Emsg("Scheduler", retc, "create time scheduler thread");
 
// If we an idle interval, schedule the idle check
//
   if (max_Workidl > 0) Schedule((XrdJob *)this, (time_t)max_Workidl+time(0));
 
// Start 1/3 of the minimum number of threads
//
   if (!(numw = min_Workers/3)) numw = 2;
   while(numw--) hireWorker(0);
 
// Unlock the data area
//
   TRACE(SCHED, "Starting with " <<num_Workers <<" workers" );
}

References XrdJob::XrdJob(), XrdSysThread::Run(), Schedule(), TRACE, TRACE_SCHED, XrdStartTSched(), and XRDSYSTHREAD_BIND.

Here is the call graph for this function:

Stats()

int XrdScheduler::Stats	(	char *	buff,
		int	blen,
		int	do_sync = 0 )

Definition at line 631 of file XrdScheduler.cc.

{
    int cnt_Jobs, cnt_JobsinQ, xam_QLength, cnt_Workers, cnt_idl;
    int cnt_TCreate, cnt_TDestroy, cnt_Limited;
    static const char statfmt[] = "<stats id=\"sched\"><jobs>%d</jobs>"
                "<inq>%d</inq><maxinq>%d</maxinq>"
                "<threads>%d</threads><idle>%d</idle>"
                "<tcr>%d</tcr><tde>%d</tde>"
                "<tlimr>%d</tlimr></stats>";
 
// If only length wanted, do so
//
   if (!buff) return sizeof(statfmt) + 16*8;
 
// Get values protected by the Dispatch lock (avoid lock if no sync needed)
//
   if (do_sync) DispatchMutex.Lock();
   cnt_idl = idl_Workers;
   if (do_sync) DispatchMutex.UnLock();
 
// Get values protected by the Scheduler lock (avoid lock if no sync needed)
//
   if (do_sync) SchedMutex.Lock();
   cnt_Workers = num_Workers;
   cnt_Jobs    = num_Jobs;
   cnt_JobsinQ = num_JobsinQ;
   xam_QLength = max_QLength;
   cnt_TCreate = num_TCreate;
   cnt_TDestroy= num_TDestroy;
   cnt_Limited = num_Limited;
   if (do_sync) SchedMutex.UnLock();
 
// Format the stats and return them
//
   return snprintf(buff, blen, statfmt, cnt_Jobs, cnt_JobsinQ, xam_QLength,
                   cnt_Workers, cnt_idl, cnt_TCreate, cnt_TDestroy,
                   cnt_Limited);
}

References max_QLength, num_Jobs, num_Limited, num_TCreate, and num_TDestroy.

TimeSched()

void XrdScheduler::TimeSched

(

)

Definition at line 674 of file XrdScheduler.cc.

{
   XrdJob *jp;
   int wtime;
 
// Continuous loop until we find some work here
//
   do {TimerMutex.Lock();
       if (TimerQueue) wtime = TimerQueue->SchedTime-time(0);
          else wtime = 60*60;
       if (wtime > 0)
          {TimerRings.Wait(wtime);
           TimerMutex.UnLock();
          } else {
           jp = TimerQueue;
           TimerQueue = jp->NextJob;
           Schedule(jp);
           TimerMutex.UnLock();
          }
       } while(1);
}

References XrdJob::XrdJob(), XrdJob::NextJob, and Schedule().

Referenced by XrdStartTSched().

Here is the call graph for this function:

Here is the caller graph for this function:

Member Data Documentation

max_QLength

int XrdScheduler::max_QLength

Definition at line 84 of file XrdScheduler.hh.

Referenced by Schedule(), Schedule(), and Stats().

num_Jobs

int XrdScheduler::num_Jobs

Definition at line 83 of file XrdScheduler.hh.

Referenced by Schedule(), Schedule(), and Stats().

num_Limited

int XrdScheduler::num_Limited

Definition at line 85 of file XrdScheduler.hh.

Referenced by Stats().

num_TCreate

int XrdScheduler::num_TCreate

Definition at line 81 of file XrdScheduler.hh.

Referenced by Stats().

num_TDestroy

int XrdScheduler::num_TDestroy

Definition at line 82 of file XrdScheduler.hh.

Referenced by Run(), and Stats().

---

The documentation for this class was generated from the following files:

• XrdScheduler.hh
• XrdScheduler.cc