34#include <sys/resource.h>
39#include <AvailabilityMacros.h>
48#define XRD_TRACE XrdTrace->
55 const char *XrdScheduler::TraceID =
"Sched";
98 int minw,
int maxw,
int maxi)
99 :
XrdJob(
"underused thread monitor"),
100 XrdTraceOld(0), WorkAvail(0,
"sched work"),
101 TimerRings(TimerMutex)
103 Boot(eP, tP, minw, maxw, maxi);
110 int minw,
int maxw,
int maxi)
111 :
XrdJob(
"underused thread monitor"),
112 XrdTraceOld(tP), WorkAvail(0,
"sched work"),
113 TimerRings(TimerMutex)
126 :
XrdJob(
"underused thread monitor"),
127 XrdTraceOld(0), WorkAvail(0,
"sched work"),
128 TimerRings(TimerMutex)
135#if ( defined(__linux__) || defined(__GNU__) ) && defined(F_DUPFD_CLOEXEC)
136 eFD = fcntl(STDERR_FILENO, F_DUPFD_CLOEXEC, 0);
138 eFD = dup(STDERR_FILENO);
139 fcntl(eFD, F_SETFD, FD_CLOEXEC);
150 if (getenv(
"XRDDEBUG") != 0) XrdTrace->What =
TRACE_SCHED;
154 Init(minw, maxw, maxi);
162 int minw,
int maxw,
int maxi)
168 Init(minw, maxw, maxi);
199 while(p && p != jp) {pp = p; p = p->
NextJob;}
206 TRACE(SCHED,
"time event " <<jp->
Comment <<
" cancelled");
220 int num_kill, num_idle;
225 {DispatchMutex.Lock(); num_idle = idl_Workers; DispatchMutex.UnLock();
226 num_kill = num_idle - min_Workers;
227 TRACE(SCHED, num_Workers <<
" threads; " <<num_idle <<
" idle");
229 {
if (num_kill > 1) num_kill = num_kill/2;
231 num_Layoffs = num_kill;
232 while(num_kill--) WorkAvail.Post();
239 if (max_Workidl > 0)
Schedule((
XrdJob *)
this, max_Workidl+time(0));
250 static int retc, ReaperStarted = 0;
256 if ((pid = fork()) < 0)
257 {XrdLog->Emsg(
"Scheduler",errno,
"fork to handle",
id);
260 if (!pid)
return pid;
266 retc = ReaperStarted;
268 ReaperMutex.UnLock();
274 0,
"Process reaper")))
275 {XrdLog->Emsg(
"Scheduler", retc,
"create reaper thread");
291#if defined(__APPLE__) && !defined(MAC_OS_X_VERSION_10_5)
292 struct timespec ts = { 1, 0 };
300 sigaddset(&Sset, SIGCHLD);
305 do {ReaperMutex.Lock();
306 tp = firstPID; ptp = 0;
308 {
do {pid = waitpid(tp->
pid, &status, WNOHANG);}
309 while (pid < 0 && errno == EINTR);
312 xtp = tp; tp = tp->
next;
313 if (ptp) ptp->
next = tp;
316 }
else {ptp = tp; tp = tp->
next;}
318 ReaperMutex.UnLock();
319#if defined(__APPLE__) && !defined(MAC_OS_X_VERSION_10_5)
321 }
while (nanosleep(&ts, 0) <= 0);
323 }
while(sigwait(&Sset, &signum) >= 0);
339 do {
do {DispatchMutex.Lock(); idl_Workers++;DispatchMutex.UnLock();
341 DispatchMutex.Lock();waiting = --idl_Workers;DispatchMutex.UnLock();
343 if ((jp = WorkFirst))
344 {
if (!(WorkFirst = jp->
NextJob)) WorkLast = 0;
345 if (num_JobsinQ) num_JobsinQ--;
346 else XrdLog->Emsg(
"Scheduler",
"Job queue count underflow!");
353 TRACE(SCHED,
"terminating thread; workers=" <<num_Workers);
365 if (!waiting) hireWorker();
367 {
TRACE(SCHED,
"running " <<jp->
Comment <<
" inq=" <<num_JobsinQ);}
386 {WorkLast->NextJob = jp;
418 {WorkLast->NextJob = jfirst;
428 num_JobsinQ += numjobs;
433 while(numjobs--) WorkAvail.Post();
453 {
TRACE(SCHED,
"scheduling " <<jp->
Comment <<
" in " <<atime-time(0) <<
" seconds");}
454 jp->SchedTime = atime;
460 while(p && p->SchedTime <= atime) {pp = p; p = p->
NextJob;}
466 else {TimerQueue = jp; TimerRings.Signal();}
492#if ( defined(__linux__) || defined(__GNU__) || (defined(__FreeBSD_kernel__) && defined(__GLIBC__)) ) && defined(RLIMIT_NPROC)
500 if ((pdFD =
open(
"/proc/sys/kernel/pid_max", O_RDONLY)) >= 0)
502 if ((rdsz =
read(pdFD, pmBuff,
sizeof(pmBuff))) > 0)
503 {rdsz = atoi(pmBuff);
504 if (rdsz < 16384) theMax = 16384;
506 theMax =
static_cast<rlim_t
>(rdsz-2000);
513 const bool setnp = (getenv(
"XRDLEAVENPROC") == 0);
518 if (setnp && !getrlimit(RLIMIT_NPROC, &rlim))
519 {
if (rlim.rlim_max == RLIM_INFINITY || rlim.rlim_max > theMax)
520 {
if (limlower || (rlim.rlim_cur != RLIM_INFINITY && rlim.rlim_cur < theMax))
521 {rlim.rlim_cur = theMax;
522 setrlimit(RLIMIT_NPROC, &rlim);
525 if (rlim.rlim_cur != rlim.rlim_max)
526 {rlim.rlim_cur = rlim.rlim_max;
527 setrlimit(RLIMIT_NPROC, &rlim);
534 if (!getrlimit(RLIMIT_NPROC, &rlim))
535 {
if (rlim.rlim_cur == RLIM_INFINITY || rlim.rlim_cur > theMax)
536 max_Workers =
static_cast<int>(theMax);
537 else max_Workers =
static_cast<int>(rlim.rlim_cur);
553 static int isSet = 0;
558 if (once && isSet) {SchedMutex.UnLock();
return;}
563 if (maxw <= 0) maxw = max_Workers;
564 if (minw < 0) minw = min_Workers;
565 if (minw > maxw) minw = maxw;
566 if (avlw < 0) avlw = maxw/4*3;
567 else if (avlw > maxw) avlw = maxw;
573 stk_Workers = maxw - avlw;
574 if (maxi >=0) max_Workidl = maxi;
589 TRACE(SCHED,
"Set min_Workers=" <<min_Workers <<
" max_Workers=" <<max_Workers);
590 TRACE(SCHED,
"Set stk_Workers=" <<stk_Workers <<
" max_Workidl=" <<max_Workidl);
604 if (getenv(
"XRDDEBUG") != 0) XrdTrace->What =
TRACE_SCHED;
605 else if (XrdTraceOld) XrdTrace->What |= XrdTraceOld->What;
611 XrdLog->Emsg(
"Scheduler", retc,
"create time scheduler thread");
615 if (max_Workidl > 0)
Schedule((
XrdJob *)
this, (time_t)max_Workidl+time(0));
619 if (!(numw = min_Workers/3)) numw = 2;
620 while(numw--) hireWorker(0);
624 TRACE(SCHED,
"Starting with " <<num_Workers <<
" workers" );
633 int cnt_Jobs, cnt_JobsinQ, xam_QLength, cnt_Workers, cnt_idl;
634 int cnt_TCreate, cnt_TDestroy, cnt_Limited;
635 static const char statfmt[] =
"<stats id=\"sched\"><jobs>%d</jobs>"
636 "<inq>%d</inq><maxinq>%d</maxinq>"
637 "<threads>%d</threads><idle>%d</idle>"
638 "<tcr>%d</tcr><tde>%d</tde>"
639 "<tlimr>%d</tlimr></stats>";
643 if (!buff)
return sizeof(statfmt) + 16*8;
647 if (do_sync) DispatchMutex.Lock();
648 cnt_idl = idl_Workers;
649 if (do_sync) DispatchMutex.UnLock();
653 if (do_sync) SchedMutex.Lock();
654 cnt_Workers = num_Workers;
656 cnt_JobsinQ = num_JobsinQ;
661 if (do_sync) SchedMutex.UnLock();
665 return snprintf(buff, blen, statfmt, cnt_Jobs, cnt_JobsinQ, xam_QLength,
666 cnt_Workers, cnt_idl, cnt_TCreate, cnt_TDestroy,
681 do {TimerMutex.Lock();
682 if (TimerQueue) wtime = TimerQueue->SchedTime-time(0);
685 {TimerRings.Wait(wtime);
703void XrdScheduler::hireWorker(
int dotrace)
711 if (num_Workers >= max_Workers)
714 XrdLog->
Emsg(
"Scheduler",
"Thread limit has been reached!");
730 {XrdLog->Emsg(
"Scheduler", retc,
"create worker thread");
734 max_Workers = num_Workers;
735 min_Workers = (max_Workers/10 ? max_Workers/10 : 1);
736 stk_Workers = max_Workers/4*3;
738 }
else if (dotrace)
TRACE(SCHED,
"Now have " <<num_Workers <<
" workers" );
745void XrdScheduler::Init(
int minw,
int maxw,
int maxi)
752 stk_Workers = maxw - (maxw/4*3);
761 WorkFirst = WorkLast = TimerQueue = 0;
768void XrdScheduler::traceExit(pid_t pid,
int status)
772 if (WIFEXITED(status))
773 {retc = WEXITSTATUS(status);
774 why =
" exited with rc=";
775 }
else if (WIFSIGNALED(status))
776 {retc = WTERMSIG(status);
777 why =
" killed with signal ";
779 why =
" changed state ";
781 TRACE(SCHED,
"Process " <<pid <<why <<retc);
XrdSysError XrdLog(0, "")
void * XrdStartWorking(void *carg)
void * XrdStartReaper(void *carg)
void * XrdStartTSched(void *carg)
#define XRDSYSTHREAD_BIND
friend class XrdScheduler
XrdJob(const char *desc="")
XrdSchedulerPID(pid_t newpid, XrdSchedulerPID *prev)
int Stats(char *buff, int blen, int do_sync=0)
void Schedule(XrdJob *jp)
void setParms(int minw, int maxw, int avlt, int maxi, int once=0)
void setNproc(const bool limlower)
pid_t Fork(const char *id)
int Emsg(const char *esfx, int ecode, const char *text1, const char *text2=0)
XrdSysError(XrdSysLogger *lp, const char *ErrPrefix="sys")
XrdSysLogger * logger(XrdSysLogger *lp=0)
XrdSysLogger(int ErrFD=STDERR_FILENO, int xrotate=1)
static int Run(pthread_t *, void *(*proc)(void *), void *arg, int opts=0, const char *desc=0)
XrdSysTrace(const char *pfx, XrdSysLogger *logp=0, int tf=0)