Mail-MIMEDefang
view release on metacpan or search on metacpan
mimedefang-multiplexor.c view on Meta::CPAN
return NULL;
}
/**********************************************************************
* %FUNCTION: usage
* %ARGUMENTS:
* None
* %RETURNS:
* Nothing (exits)
* %DESCRIPTION:
* Prints usage information
***********************************************************************/
static void
usage(void)
{
fprintf(stderr, "mimedefang-multiplexor version %s\n", VERSION);
fprintf(stderr, "Usage: mimedefang-multiplexor [options]\n");
fprintf(stderr, "Options:\n");
fprintf(stderr, " -h -- Print usage info and exit\n");
fprintf(stderr, " -v -- Print version and exit\n");
fprintf(stderr, " -t filename -- Log statistics to filename\n");
fprintf(stderr, " -p filename -- Write process-ID in filename\n");
fprintf(stderr, " -o file -- Use specified file as a lock file\n");
fprintf(stderr, " -T -- Log statistics to syslog\n");
fprintf(stderr, " -u -- Flush stats file after each write\n");
fprintf(stderr, " -Z -- Accept and process status updates from busy workers\n");
fprintf(stderr, " -U username -- Run as username, not root\n");
fprintf(stderr, " -m minWorkers -- Minimum number of workers\n");
fprintf(stderr, " -x maxWorkers -- Maximum number of workers\n");
fprintf(stderr, " -y recipokPerDom -- Maximum concurrent recipoks per domain\n");
fprintf(stderr, " -r maxRequests -- Maximum number of requests per worker\n");
fprintf(stderr, " -V maxLifetime -- Maximum lifetime of a worker in seconds\n");
fprintf(stderr, " -i idleTime -- Idle time (seconds) for killing excess workers\n");
fprintf(stderr, " -b busyTime -- Busy time (seconds) for killing hung workers\n");
fprintf(stderr, " -c cmdTime -- Request/reply transmission timeout (seconds)\n");
fprintf(stderr, " -w waitTime -- How long to wait between worker activations (seconds)\n");
fprintf(stderr, " -W waitTime -- Absolute minimum to wait between worker activations\n");
fprintf(stderr, " -z dir -- Spool directory\n");
fprintf(stderr, " -s sock -- UNIX-domain socket for communication\n");
fprintf(stderr, " -a u_sock -- Socket for unprivileged communication\n");
fprintf(stderr, " -f /dir/filter -- Specify full path of filter program\n");
fprintf(stderr, " -d -- Debug events in /var/log/mdefang-event-debug.log\n");
fprintf(stderr, " -l -- Log events with syslog\n");
#ifdef HAVE_SETRLIMIT
fprintf(stderr, " -R size -- Limit RSS to size kB (if supported on your OS)\n");
fprintf(stderr, " -M size -- Limit memory address space to size kB\n");
#endif
fprintf(stderr, " -L interval -- Log worker status every interval seconds\n");
fprintf(stderr, " -S facility -- Set syslog(3) facility\n");
fprintf(stderr, " -N sock -- Listen for Sendmail map requests on sock\n");
fprintf(stderr, " -O sock -- Listen for notification requests on sock\n");
fprintf(stderr, " -q size -- Size of request queue (default 0)\n");
fprintf(stderr, " -Q timeout -- Timeout for queued requests\n");
fprintf(stderr, " -I backlog -- 'backlog' argument for listen on multiplexor socket\n");
fprintf(stderr, " -D -- Do not become a daemon (stay in foreground)\n");
fprintf(stderr, " -X interval -- Run a 'tick' request every interval seconds\n");
fprintf(stderr, " -P n -- Run 'n' parallel tick requests\n");
fprintf(stderr, " -Y label -- Set syslog label to 'label'\n");
fprintf(stderr, " -G -- Make sockets group-writable\n");
fprintf(stderr, " -k -- Enable adaptive autoscaling of the worker pool\n");
#ifdef EMBED_PERL
fprintf(stderr, " -E -- Use embedded Perl interpreter\n");
#endif
exit(EXIT_FAILURE);
}
static int
set_sigchld_handler(void)
{
struct sigaction act;
/* Set signal handler for SIGCHLD */
act.sa_handler = childHandler;
sigemptyset(&act.sa_mask);
act.sa_flags = SA_NOCLDSTOP | SA_RESTART;
return sigaction(SIGCHLD, &act, NULL);
}
/**********************************************************************
* %FUNCTION: main
* %ARGUMENTS:
* argc, argv -- usual suspects
* %RETURNS:
* Nothing -- runs in an infinite loop
* %DESCRIPTION:
* Main program
***********************************************************************/
int
main(int argc, char *argv[], char **env)
{
int i;
int sock, unpriv_sock;
int c;
int n;
int pidfile_fd = -1;
int lockfile_fd = -1;
char *user = NULL;
char *options;
int facility = LOG_MAIL;
int kidpipe[2] = {-1, -1};
char kidmsg[256];
time_t now;
mode_t socket_umask = 077;
mode_t file_umask = 077;
EventSelector *es;
struct sigaction act;
struct timeval t;
struct passwd *pw = NULL;
int nodaemon = 0;
/* Record program start time */
TimeOfProgramStart = time(NULL);
Env = env;
/* Paranoia time */
umask(077);
mimedefang-multiplexor.c view on Meta::CPAN
Settings.maxAS = 0;
Settings.logStatusInterval = 0;
Settings.requestQueueSize = 0;
Settings.requestQueueTimeout = 30;
Settings.listenBacklog = -1;
Settings.useEmbeddedPerl = 0;
Settings.notifySock = NULL;
Settings.tick_interval = 0;
Settings.num_ticks = 1;
Settings.mapSock = NULL;
Settings.wantStatusReports = 0;
Settings.debugWorkerScheduling = 0;
Settings.autoscaling = 0;
Settings.autoscaleInterval = 15;
Settings.scaleOutBusyRatio = 0.80;
Settings.scaleInBusyRatio = 0.30;
Settings.scaleOutCooldown = 5;
Settings.scaleInCooldown = 30;
Settings.emaAlpha = 0.25;
#ifndef HAVE_SETRLIMIT
options = "GAa:Tt:um:x:y:r:i:b:c:s:hdlf:p:o:w:F:W:U:S:q:Q:I:DEO:X:Y:N:vZP:z:V:k";
#else
options = "GAa:Tt:um:x:y:r:i:b:c:s:hdlf:p:o:w:F:W:U:S:q:Q:L:R:M:I:DEO:X:Y:N:vZP:z:V:k";
#endif
while((c = getopt(argc, argv, options)) != -1) {
switch(c) {
case 'G':
socket_umask = 007;
file_umask = 027;
break;
case 'A':
Settings.debugWorkerScheduling = 1;
break;
case 'k':
Settings.autoscaling = 1;
break;
case 'z':
Settings.spoolDir = strdup(optarg);
if (!Settings.spoolDir) {
fprintf(stderr, "%s: Out of memory\n", argv[0]);
exit(EXIT_FAILURE);
}
break;
case 'Z':
Settings.wantStatusReports = 1;
break;
case 'a':
Settings.unprivSockName = strdup(optarg);
if (!Settings.unprivSockName) {
fprintf(stderr, "%s: Out of memory\n", argv[0]);
exit(EXIT_FAILURE);
}
break;
case 'v':
printf("mimedefang-multiplexor version %s\n", VERSION);
exit(EXIT_SUCCESS);
case 'E':
#ifdef EMBED_PERL
Settings.useEmbeddedPerl = 1;
#else
fprintf(stderr, "mimedefang-multiplexor compiled without support for embedded perl. Ignoring -E flag.\n");
#endif
break;
case 'D':
nodaemon = 1;
break;
case 'P':
if (sscanf(optarg, "%d", &n) != 1) usage();
if (n < 1) {
n = 1;
} else if (n > 30) {
n = 30;
}
Settings.num_ticks = n;
break;
case 'I':
if (sscanf(optarg, "%d", &n) != 1) usage();
if (n < 5) {
n = 5;
} else if (n > 200) {
n = 200;
}
Settings.listenBacklog = n;
break;
case 'q':
if (sscanf(optarg, "%d", &n) != 1) usage();
if (n <= 0) {
n = 0;
} else if (n > MAX_QUEUE_SIZE) {
fprintf(stderr, "%s: Request queue size %d too big (%d max)\n",
argv[0], n, MAX_QUEUE_SIZE);
exit(EXIT_FAILURE);
}
Settings.requestQueueSize = n;
break;
case 'X':
if (sscanf(optarg, "%d", &n) != 1) usage();
if (n < 0) {
n = 0;
}
Settings.tick_interval = n;
break;
case 'Q':
if (sscanf(optarg, "%d", &n) != 1) usage();
if (n <= 1) {
n = 1;
} else if (n > 600) {
n = 600;
}
Settings.requestQueueTimeout = n;
break;
case 'S':
facility = find_syslog_facility(optarg);
if (facility < 0) {
mimedefang-multiplexor.c view on Meta::CPAN
for (i=0; i<CLOSEFDS; i++) {
/* Don't close stdin/stdout/stderr if we are not a daemon */
if (nodaemon && i < 3) {
continue;
}
if (i == kidpipe[0] || i == kidpipe[1] || i == lockfile_fd || i == unpriv_sock || i == sock || i == Pipe[0] || i == Pipe[1]) continue;
(void) close(i);
}
/* Direct stdin/stdout/stderr to /dev/null if we are a daemon */
if (!nodaemon) {
open("/dev/null", O_RDWR);
open("/dev/null", O_RDWR);
open("/dev/null", O_RDWR);
}
/* Syslog if required */
if (Settings.syslog_label) {
openlog(Settings.syslog_label, LOG_PID|LOG_NDELAY, facility);
} else {
openlog("mimedefang-multiplexor", LOG_PID|LOG_NDELAY, facility);
}
/* Keep track of our pid */
ParentPid = getpid();
/* Set up SIGHUP handler */
act.sa_handler = hupHandler;
sigemptyset(&act.sa_mask);
act.sa_flags = SA_RESTART;
if (sigaction(SIGHUP, &act, NULL) < 0) {
REPORT_FAILURE("sigaction failed - exiting.");
if (pidfile) unlink(pidfile);
if (lockfile) unlink(lockfile);
exit(EXIT_FAILURE);
}
/* Set up SIGINT handler */
act.sa_handler = intHandler;
sigemptyset(&act.sa_mask);
act.sa_flags = SA_RESTART;
if (sigaction(SIGINT, &act, NULL) < 0) {
REPORT_FAILURE("sigaction failed - exiting.");
if (pidfile) unlink(pidfile);
if (lockfile) unlink(lockfile);
exit(EXIT_FAILURE);
}
if (DOLOG) {
syslog(LOG_INFO, "started; minWorkers=%d, maxWorkers=%d, maxRequests=%d, maxLifetime=%d, maxIdleTime=%d, busyTimeout=%d, clientTimeout=%d",
Settings.minWorkers,
Settings.maxWorkers,
Settings.maxRequests,
Settings.maxLifetime,
Settings.maxIdleTime,
Settings.busyTimeout,
Settings.clientTimeout);
}
/* Init Perl interpreter */
#ifdef EMBED_PERL
if (Settings.useEmbeddedPerl) {
init_embedded_interpreter(argc, argv, env);
if (make_embedded_interpreter(Settings.progPath, Settings.subFilter,
Settings.wantStatusReports, Env) < 0) {
syslog(LOG_ERR, "Could not initialize embedded Perl interpreter -- falling back to old method.");
Settings.useEmbeddedPerl = 0;
} else {
if (DOLOG) {
syslog(LOG_INFO, "Initialized embedded Perl interpreter");
}
}
}
#endif
/* Set signal handler for SIGCHLD */
if (set_sigchld_handler() < 0) {
REPORT_FAILURE("sigaction failed - exiting.");
#ifdef EMBED_PERL
if (Settings.useEmbeddedPerl) {
term_embedded_interpreter();
deinit_embedded_interpreter();
}
#endif
if (pidfile) unlink(pidfile);
if (lockfile) unlink(lockfile);
exit(EXIT_FAILURE);
}
/* Open stats file */
statsReopenFile();
/* Kick off the starting of workers */
bringWorkersUpToMin(es, 0, 0, NULL);
/* Set up a timer handler to check for idle timeouts */
t.tv_usec = 0;
t.tv_sec = Settings.maxIdleTime;
Event_AddTimerHandler(es, t, handleIdleTimeout, NULL);
/* Set up adaptive autoscaling timer if enabled (-k flag) */
if (Settings.autoscaling) {
t.tv_usec = 0;
t.tv_sec = Settings.autoscaleInterval;
Event_AddTimerHandler(es, t, handleAutoscale, NULL);
syslog(LOG_INFO,
"Autoscaling enabled: interval=%ds scale_out=%.0f%% scale_in=%.0f%%",
Settings.autoscaleInterval,
Settings.scaleOutBusyRatio * 100.0,
Settings.scaleInBusyRatio * 100.0);
}
/* Set up a timer handler to log status, if desired */
if (Settings.logStatusInterval) {
t.tv_usec = 0;
t.tv_sec = Settings.logStatusInterval;
Event_AddTimerHandler(es, t, doStatusLog, NULL);
}
if (DebugEvents) {
Event_EnableDebugging("/var/log/mdefang-event-debug.log");
}
/* Set signal handler for SIGTERM */
signal(SIGTERM, sigterm);
/* Do notify socket */
if (Settings.notifySock) {
umask(socket_umask);
make_notifier_socket(es, Settings.notifySock);
umask(file_umask);
}
if (Settings.mapSock) {
umask(socket_umask);
sock = make_listening_socket(Settings.mapSock, Settings.listenBacklog, 0);
umask(file_umask);
if (sock >= 0) {
if(set_cloexec(sock) < 0) {
syslog(LOG_ERR, "Could not set FD_CLOEXEC option on socket");
mimedefang-multiplexor.c view on Meta::CPAN
}
if (len == 5 && !strcmp(buf, "hload")) {
doHourlyLoad(es, fd, SCAN_CMD);
return;
}
if (len == 13 && !strcmp(buf, "hload-relayok")) {
doHourlyLoad(es, fd, RELAYOK_CMD);
return;
}
if (len == 14 && !strcmp(buf, "hload-senderok")) {
doHourlyLoad(es, fd, SENDEROK_CMD);
return;
}
if (len == 13 && !strcmp(buf, "hload-recipok")) {
doHourlyLoad(es, fd, RECIPOK_CMD);
return;
}
if (len == 5 && !strcmp(buf, "histo")) {
doHistogram(es, fd);
return;
}
if (len == 4 && !strcmp(buf, "msgs")) {
snprintf(answer, sizeof(answer), "%d\n", NumMsgsProcessed);
reply_to_mimedefang(es, fd, answer);
return;
}
/* We have to keep the old command for backward-compatibility */
if (len > 10 && !strncmp(buf, "slaveinfo ", 10)) {
doWorkerInfo(es, fd, buf);
return;
}
if (len > 11 && !strncmp(buf, "workerinfo ", 11)) {
doWorkerInfo(es, fd, buf);
return;
}
/* This is an awful hack used by watch-multiple-mimedefangs.tcl.
We handle it here so we don't have to waste a worker */
if (len == 19 && !strcmp(buf, "foo_no_such_command")) {
reply_to_mimedefang(es, fd, "error: Unknown command\n");
return;
}
/* Remaining commands are privileged */
if (data) {
reply_to_mimedefang(es, fd, "error: Attempt to use privileged command on unprivileged socket\n");
return;
}
if (len == 6 && !strcmp(buf, "reread")) {
newGeneration();
notify_listeners(es, "R\n");
#ifndef SAFE_EMBED_PERL
if (Settings.useEmbeddedPerl) {
reply_to_mimedefang(es, fd, "Cannot destroy and recreate an embedded Perl interpreter safely on this platform. Filter rules will NOT be reread\n");
return;
}
#endif
reply_to_mimedefang(es, fd, "Forced reread of filter rules\n");
return;
}
if (len == 9 && !strcmp(buf, "autoscale")) {
doAutoscaleStatus(es, fd);
return;
}
if (len > 5 && !strncmp(buf, "scan ", 5)) {
doScan(es, fd, buf);
return;
}
/* Any command other than "scan" is handled generically. */
doWorkerCommand(es, fd, buf);
return;
}
static int
worker_request_age(Worker *s) {
if (s->firstReqTime == (time_t) -1) {
return -1;
}
return (int) (time(NULL) - s->firstReqTime);
}
static int
worker_age(Worker *s) {
if (s->activationTime == (time_t) -1) {
return -1;
}
return (int) (time(NULL) - s->activationTime);
}
/**********************************************************************
* %FUNCTION: doWorkerInfo
* %ARGUMENTS:
* es -- event selector
* fd -- connection to MIMEDefang
* cmd -- workerinfo command
* %RETURNS:
* Nothing
* %DESCRIPTION:
* Returns detailed information about a specific worker
***********************************************************************/
static void
doWorkerInfo(EventSelector *es, int fd, char *cmd)
{
int workerno;
char buf[1024];
Worker *s;
if (sscanf(cmd+10, "%d", &workerno) != 1) {
reply_to_mimedefang(es, fd, "error: Invalid worker number\n");
mimedefang-multiplexor.c view on Meta::CPAN
/* Every STOPPED->running transition is effectively a scale-out
* event. Log it consistently and update LastScaleOut so the
* scale-in cooldown is honoured against lazy-spawn activations
* (findFreeWorker -> activateWorker) just as it is for the
* explicit EMA / queued-request paths. */
if (Settings.autoscaling) {
LastScaleOut = (now != (time_t) 0) ? now : time(NULL);
syslog(LOG_INFO,
"Autoscale: scaled out to %d workers (ema_busy=%.2f queued=%d)",
NUM_RUNNING_WORKERS, EMABusyRatio, NumQueuedRequests);
}
return s->pid;
}
/* In the child */
/* Reset signal-handling dispositions */
signal(SIGTERM, SIG_DFL);
signal(SIGCHLD, SIG_DFL);
signal(SIGHUP, SIG_DFL);
signal(SIGINT, SIG_DFL);
sigemptyset(&sigs);
sigprocmask(SIG_SETMASK, &sigs, NULL);
/* Set resource limits */
#ifdef HAVE_SETRLIMIT
limit_mem_usage(Settings.maxRSS, Settings.maxAS);
#endif
/* Close unneeded file descriptors */
closelog();
close(pin[1]);
close(pout[0]);
close(perr[0]);
dup2(pin[0], STDIN_FILENO);
dup2(pout[1], STDOUT_FILENO);
dup2(perr[1], STDERR_FILENO);
if (pin[0] != STDIN_FILENO) close(pin[0]);
if (pout[1] != STDOUT_FILENO) close(pout[1]);
if (perr[1] != STDERR_FILENO) close(perr[1]);
if (Settings.wantStatusReports) {
dup2(pstatus[1], STDERR_FILENO+1);
if (pstatus[1] != STDERR_FILENO+1) close(pstatus[1]);
}
if (!Settings.wantStatusReports) (void) close(STDERR_FILENO+1);
for (i=STDERR_FILENO+2; i<1024; i++) {
(void) close(i);
}
pname = strrchr(Settings.progPath, '/');
if (pname) {
pname++;
} else {
pname = Settings.progPath;
}
#ifdef EMBED_PERL
if (Settings.useEmbeddedPerl) {
run_embedded_filter();
term_embedded_interpreter();
deinit_embedded_interpreter();
exit(EXIT_SUCCESS);
}
#endif
if (Settings.wantStatusReports) {
sarg = "-serveru";
} else {
sarg = "-server";
}
if (Settings.subFilter) {
execl(Settings.progPath, pname, "-f", Settings.subFilter, sarg, NULL);
} else {
execl(Settings.progPath, pname, sarg, NULL);
}
_exit(EXIT_FAILURE);
}
/**********************************************************************
* %FUNCTION: checkWorkerForExpiry
* %ARGUMENTS:
* s -- a worker to check
* %RETURNS:
* Nothing
* %DESCRIPTION:
* If the worker has served too many requests, it is killed.
***********************************************************************/
static void
checkWorkerForExpiry(Worker *s)
{
/* If there is a queued request, don't terminate worker just yet. Allow
it to go up to triple maxRequests. Yes, this is a horrible hack. */
if (s->numRequests < Settings.maxRequests * 3) {
if (handle_queued_request()) {
return;
}
}
if (s->numRequests >= Settings.maxRequests) {
char reason[200];
snprintf(reason, sizeof(reason), "Worker has processed %d requests",
s->numRequests);
killWorker(s, reason);
} else if (Settings.maxLifetime > 0 && worker_request_age(s) > Settings.maxLifetime) {
char reason[200];
snprintf(reason, sizeof(reason), "Worker has exceeded maximum lifetime of %d seconds", Settings.maxLifetime);
killWorker(s, reason);
} else if (s->generation < Generation) {
killWorker(s, "New generation -- forcing reread of filter rules");
}
}
/**********************************************************************
* %FUNCTION: killWorker
* %ARGUMENTS:
* s -- a worker to kill
* reason -- reason worker is being killed
* %RETURNS:
* Nothing
mimedefang-multiplexor.c view on Meta::CPAN
* sig -- signal number
* %RETURNS:
* Nothing
* %DESCRIPTION:
* Called when SIGTERM received -- kills workers and exits
***********************************************************************/
static void
sigterm(int sig)
{
int i, j, oneleft;
/* Only the parent process should handle SIGTERM */
if (ParentPid != getpid()) {
syslog(LOG_WARNING, "Child process received SIGTERM before signal disposition could be reset! Exiting!");
exit(EXIT_FAILURE);
}
if (pidfile) {
unlink(pidfile);
}
if (lockfile) {
unlink(lockfile);
}
if (DOLOG) {
if (sig) {
syslog(LOG_INFO, "Received SIGTERM: Stopping workers and terminating");
}
}
/* Remove our socket so we don't get any more requests */
if (Settings.sockName) {
(void) remove(Settings.sockName);
}
/* Hack...*/
if (Settings.unprivSockName && (Settings.unprivSockName[0] == '/')) {
(void) remove(Settings.unprivSockName);
}
/* First, close descriptors to force EOF on STDIN; then wait up to 10
seconds before sending SIGTERM */
for (i=0; i<Settings.maxWorkers; i++) {
if (AllWorkers[i].pid != (pid_t) -1) {
kill(AllWorkers[i].pid, SIGCONT);
close(AllWorkers[i].workerStdin);
AllWorkers[i].workerStdin = -1;
}
}
/* Wait up to 10 seconds for workers to exit; then kill with SIGTERM */
for (j=0; j<10; j++) {
reapTerminatedWorkers(1);
oneleft = 0;
for (i=0; i<Settings.maxWorkers; i++) {
if (AllWorkers[i].pid != (pid_t) -1) {
oneleft = 1;
break;
}
}
if (!oneleft) {
#ifdef EMBED_PERL
if (Settings.useEmbeddedPerl) {
term_embedded_interpreter();
deinit_embedded_interpreter();
}
#endif
exit(EXIT_SUCCESS);
}
if (j != 9) {
sleep(1);
}
}
syslog(LOG_INFO, "Still some workers alive: Sending SIGTERM");
/* Still some workers. SIGTERM them */
for (i=0; i<Settings.maxWorkers; i++) {
if (AllWorkers[i].pid != (pid_t) -1) {
kill(AllWorkers[i].pid, SIGCONT);
kill(AllWorkers[i].pid, SIGTERM);
}
}
/* Wait up to 10 seconds for workers to exit; then kill with SIGKILL */
for (j=0; j<10; j++) {
reapTerminatedWorkers(1);
oneleft = 0;
for (i=0; i<Settings.maxWorkers; i++) {
if (AllWorkers[i].pid != (pid_t) -1) {
oneleft = 1;
break;
}
}
if (!oneleft) {
#ifdef EMBED_PERL
if (Settings.useEmbeddedPerl) {
term_embedded_interpreter();
deinit_embedded_interpreter();
}
#endif
exit(EXIT_SUCCESS);
}
if (j != 9) {
sleep(1);
}
}
syslog(LOG_INFO, "Still some workers alive: Sending SIGKILL");
/* Kill with SIGKILL */
for (i=0; i<Settings.maxWorkers; i++) {
if (AllWorkers[i].pid != (pid_t) -1) {
kill(AllWorkers[i].pid, SIGCONT);
kill(AllWorkers[i].pid, SIGKILL);
}
}
#ifdef EMBED_PERL
if (Settings.useEmbeddedPerl) {
term_embedded_interpreter();
deinit_embedded_interpreter();
}
#endif
exit(EXIT_SUCCESS);
}
/**********************************************************************
* %FUNCTION: statsReopenFile
* %ARGUMENTS:
* None
* %RETURNS:
* Nothing
* %DESCRIPTION:
* Opens or re-opens statistics file.
***********************************************************************/
static void
statsReopenFile(void)
{
if (!Settings.statsFile) return;
if (Settings.statsFP) {
if (fclose(Settings.statsFP) == EOF) {
syslog(LOG_ERR, "Failed to close stats file: %m");
}
Settings.statsFP = NULL;
}
Settings.statsFP = fopen(Settings.statsFile, "a");
if (!Settings.statsFP) {
syslog(LOG_ERR, "Could not open stats file %s: %m",
Settings.statsFile);
} else {
if(set_cloexec(fileno(Settings.statsFP)) < 0) {
syslog(LOG_ERR, "Could not set FD_CLOEXEC option on socket");
}
}
}
/**********************************************************************
* %FUNCTION: statsLog
* %ARGUMENTS:
* event -- name of event to put in log
* workerno -- worker involved in this event, or -1 if not worker-specific
* fmt -- "printf" format string with extra args to add to line
* %RETURNS:
* Nothing
* %DESCRIPTION:
* Logs an event to the stats file
***********************************************************************/
static void
statsLog(char const *event, int workerno, char const *fmt, ...)
{
struct timeval now;
time_t tnow;
struct tm *t;
char tbuf[64];
char statbuf[1024];
va_list ap;
int ms;
if (!Settings.statsFP && !Settings.statsToSyslog) return;
gettimeofday(&now, NULL);
tnow = (time_t) now.tv_sec;
t = localtime(&tnow);
ms = now.tv_usec / 1000;
strftime(tbuf, sizeof(tbuf), "%d/%m/%Y:%H:%M:%S", t);
snprintf(statbuf, sizeof(statbuf),
"%s %lu.%03d %s worker=%d nworkers=%d nbusy=%d",
tbuf, (unsigned long) tnow, ms, event, workerno,
NUM_RUNNING_WORKERS, WorkerCount[STATE_BUSY]);
statbuf[sizeof(statbuf)-1] = 0;
if (fmt) {
int len = strlen(statbuf);
statbuf[len] = ' ';
va_start(ap, fmt);
vsnprintf(statbuf+len+1, sizeof(statbuf)-len-1, fmt, ap);
va_end(ap);
statbuf[sizeof(statbuf)-1] = 0;
}
if (Settings.statsFP) {
fprintf(Settings.statsFP, "%s\n", statbuf);
if (Settings.flushStats) fflush(Settings.statsFP);
}
if (Settings.statsToSyslog) {
/* Chop off the date, because it's redundant. The date is
always "dd/mm/YYYY:hh:mm:ss " = 20 characters long. */
syslog(LOG_INFO, "stats %s", statbuf + 20);
}
}
/**********************************************************************
* %FUNCTION: newGeneration
* %ARGUMENTS:
* None
* %RETURNS:
* Nothing
* %DESCRIPTION:
* Kills all running-but-idle workers and increments Generation. This
* causes running-but-busy workers to terminate at the earliest opportunity.
* Use this if you change the filter file and want to restart the workers.
***********************************************************************/
static void
newGeneration(void)
{
Generation++;
#ifdef EMBED_PERL
if (Settings.useEmbeddedPerl) {
if (make_embedded_interpreter(Settings.progPath,
Settings.subFilter,
Settings.wantStatusReports, Env) < 0) {
syslog(LOG_ERR, "Error creating embedded Perl interpreter: Reverting to non-embedded interpreter!");
Settings.useEmbeddedPerl = 0;
} else {
if (DOLOG) {
syslog(LOG_INFO, "Re-initialized embedded Perl interpreter");
}
}
}
#endif
/* Reset SIGCHLD handler in case some Perl code has monkeyed with it */
set_sigchld_handler();
while(Workers[STATE_IDLE]) {
killWorker(Workers[STATE_IDLE],
"Forcing reread of filter rules");
}
}
/**********************************************************************
* %FUNCTION: handleAutoscale
* %ARGUMENTS:
* es -- event selector
* fd, flags, data -- ignored
* %RETURNS:
* Nothing
* %DESCRIPTION:
* Called periodically when autoscaling is enabled (-k flag).
* Uses an exponential moving average (EMA) of the busy-worker ratio
* to make scale-out and scale-in decisions with AIMD-style cooldowns.
* Scale out when the EMA exceeds scaleOutBusyRatio; scale in when it
* falls below scaleInBusyRatio and both cooldown timers have elapsed.
***********************************************************************/
static void
handleAutoscale(EventSelector *es,
int fd,
unsigned int flags,
void *data)
{
time_t now = time(NULL);
struct timeval t;
int nRunning = NUM_RUNNING_WORKERS;
int nBusy = WorkerCount[STATE_BUSY];
double rawBusy = (nRunning > 0) ? (double)nBusy / nRunning : 0.0;
char reason[128];
Worker *s;
/* Update exponential moving average of busy ratio */
EMABusyRatio = EMABusyRatio * (1.0 - Settings.emaAlpha)
+ rawBusy * Settings.emaAlpha;
/* Scale OUT: pool is saturated */
if (EMABusyRatio > Settings.scaleOutBusyRatio
&& nRunning < Settings.maxWorkers
&& (now - LastScaleOut) > (time_t)Settings.scaleOutCooldown) {
s = Workers[STATE_STOPPED];
( run in 1.478 second using v1.01-cache-2.11-cpan-3fabe0161c3 )