#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	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 45 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")
{
   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 108 of file XrdScheduler.cc.

              : XrdJob("underused thread monitor"),
                XrdTraceOld(tP), WorkAvail(0, "sched work")
{
 
// 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 123 of file XrdScheduler.cc.

              : XrdJob("underused thread monitor"),
                XrdTraceOld(0), WorkAvail(0, "sched work")
{
   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(), fcntl(), Logger, and TRACE_SCHED.

Here is the call graph for this function:

◆ ~XrdScheduler()

XrdScheduler::~XrdScheduler ( )

Definition at line 218 of file XrdScheduler.cc.

219{

220}

Member Function Documentation

◆ Active()

int XrdScheduler::Active ( )

inline

Definition at line 49 of file XrdScheduler.hh.

49{return num_Workers - idl_Workers + num_JobsinQ;}

◆ Cancel()

void XrdScheduler::Cancel ( XrdJob * jp )

Definition at line 226 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 53 of file XrdScheduler.hh.

53 {return num_Workers < stk_Workers

54 || (num_Workers-idl_Workers) < stk_Workers;}

◆ DoIt()

void XrdScheduler::DoIt ( )

virtual

Implements XrdJob.

Definition at line 256 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 286 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 324 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, and TRACING.

Referenced by XrdStartReaper().

Here is the caller graph for this function:

◆ Run()

void XrdScheduler::Run ( )

Definition at line 370 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 445 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 414 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(), DoIt(), XrdXrootdCallBack::Done(), mainAccept(), mainAdmin(), setParms(), 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 480 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:

◆ setParms()

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

Definition at line 515 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 561 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 595 of file XrdScheduler.cc.

{
    int cnt_Jobs, cnt_JobsinQ, xam_QLength, cnt_Workers, cnt_idl;
    int cnt_TCreate, cnt_TDestroy, cnt_Limited;
    static 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 638 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)
          {TimerMutex.UnLock();
           TimerRings.Wait(wtime);
          } 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 81 of file XrdScheduler.hh.

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

◆ num_Jobs

int XrdScheduler::num_Jobs

Definition at line 80 of file XrdScheduler.hh.

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

◆ num_Limited

int XrdScheduler::num_Limited

Definition at line 82 of file XrdScheduler.hh.

Referenced by Stats().

◆ num_TCreate

int XrdScheduler::num_TCreate

Definition at line 78 of file XrdScheduler.hh.

Referenced by Stats().

◆ num_TDestroy

int XrdScheduler::num_TDestroy

Definition at line 79 of file XrdScheduler.hh.

Referenced by Run(), and Stats().

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

Public Member Functions

Public Attributes

Detailed Description

Constructor & Destructor Documentation

◆ XrdScheduler() [1/3]

◆ XrdScheduler() [2/3]

◆ XrdScheduler() [3/3]

◆ ~XrdScheduler()

Member Function Documentation

◆ Active()

◆ Cancel()

◆ canStick()

◆ DoIt()

◆ Fork()

◆ Reaper()

◆ Run()

◆ Schedule() [1/3]

◆ Schedule() [2/3]

◆ Schedule() [3/3]

◆ setParms()

◆ Start()

◆ Stats()

◆ TimeSched()

Member Data Documentation

◆ max_QLength

◆ num_Jobs

◆ num_Limited

◆ num_TCreate

◆ num_TDestroy