/*
* Copyright (c) 1998-2006 Sendmail, Inc. and its suppliers.
* All rights reserved.
* Copyright (c) 1983, 1995-1997 Eric P. Allman. All rights reserved.
* Copyright (c) 1988, 1993
* The Regents of the University of California. All rights reserved.
*
* By using this file, you agree to the terms and conditions set
* forth in the LICENSE file which can be found at the top level of
* the sendmail distribution.
*
*/
#include <sendmail.h>
#include <sm/sem.h>
SM_RCSID("@(#)$Id: queue.c,v 8.954 2006/04/22 01:07:00 ca Exp $")
#include <dirent.h>
# define RELEASE_QUEUE (void) 0
# define ST_INODE(st) (st).st_ino
# define sm_file_exists(errno) ((errno) == EEXIST)
# if HASFLOCK && defined(O_EXLOCK)
# define SM_OPEN_EXLOCK 1
# define TF_OPEN_FLAGS (O_CREAT|O_WRONLY|O_EXCL|O_EXLOCK)
# else /* HASFLOCK && defined(O_EXLOCK) */
# define TF_OPEN_FLAGS (O_CREAT|O_WRONLY|O_EXCL)
# endif /* HASFLOCK && defined(O_EXLOCK) */
#ifndef SM_OPEN_EXLOCK
# define SM_OPEN_EXLOCK 0
#endif /* ! SM_OPEN_EXLOCK */
/*
** Historical notes:
** QF_VERSION == 4 was sendmail 8.10/8.11 without _FFR_QUEUEDELAY
** QF_VERSION == 5 was sendmail 8.10/8.11 with _FFR_QUEUEDELAY
** QF_VERSION == 6 was sendmail 8.12 without _FFR_QUEUEDELAY
** QF_VERSION == 7 was sendmail 8.12 with _FFR_QUEUEDELAY
** QF_VERSION == 8 is sendmail 8.13
*/
#define QF_VERSION 8 /* version number of this queue format */
static char queue_letter __P((ENVELOPE *, int));
static bool quarantine_queue_item __P((int, int, ENVELOPE *, char *));
/* Naming convention: qgrp: index of queue group, qg: QUEUEGROUP */
/*
** Work queue.
*/
struct work
{
char *w_name; /* name of control file */
char *w_host; /* name of recipient host */
bool w_lock; /* is message locked? */
bool w_tooyoung; /* is it too young to run? */
long w_pri; /* priority of message, see below */
time_t w_ctime; /* creation time */
time_t w_mtime; /* modification time */
int w_qgrp; /* queue group located in */
int w_qdir; /* queue directory located in */
struct work *w_next; /* next in queue */
};
typedef struct work WORK;
static WORK *WorkQ; /* queue of things to be done */
static int NumWorkGroups; /* number of work groups */
static time_t Current_LA_time = 0;
/* Get new load average every 30 seconds. */
#define GET_NEW_LA_TIME 30
#define SM_GET_LA(now) \
do \
{ \
now = curtime(); \
if (Current_LA_time < now - GET_NEW_LA_TIME) \
{ \
sm_getla(); \
Current_LA_time = now; \
} \
} while (0)
/*
** DoQueueRun indicates that a queue run is needed.
** Notice: DoQueueRun is modified in a signal handler!
*/
static bool volatile DoQueueRun; /* non-interrupt time queue run needed */
/*
** Work group definition structure.
** Each work group contains one or more queue groups. This is done
** to manage the number of queue group runners active at the same time
** to be within the constraints of MaxQueueChildren (if it is set).
** The number of queue groups that can be run on the next work run
** is kept track of. The queue groups are run in a round robin.
*/
struct workgrp
{
int wg_numqgrp; /* number of queue groups in work grp */
int wg_runners; /* total runners */
int wg_curqgrp; /* current queue group */
QUEUEGRP **wg_qgs; /* array of queue groups */
int wg_maxact; /* max # of active runners */
time_t wg_lowqintvl; /* lowest queue interval */
int wg_restart; /* needs restarting? */
int wg_restartcnt; /* count of times restarted */
};
typedef struct workgrp WORKGRP;
static WORKGRP volatile WorkGrp[MAXWORKGROUPS + 1]; /* work groups */
#if SM_HEAP_CHECK
static SM_DEBUG_T DebugLeakQ = SM_DEBUG_INITIALIZER("leak_q",
"@(#)$Debug: leak_q - trace memory leaks during queue processing $");
#endif /* SM_HEAP_CHECK */
/*
** We use EmptyString instead of "" to avoid
** 'zero-length format string' warnings from gcc
*/
static const char EmptyString[] = "";
static void grow_wlist __P((int, int));
static int multiqueue_cache __P((char *, int, QUEUEGRP *, int, unsigned int *));
static int gatherq __P((int, int, bool, bool *, bool *));
static int sortq __P((int));
static void printctladdr __P((ADDRESS *, SM_FILE_T *));
static bool readqf __P((ENVELOPE *, bool));
static void restart_work_group __P((int));
static void runner_work __P((ENVELOPE *, int, bool, int, int));
static void schedule_queue_runs __P((bool, int, bool));
static char *strrev __P((char *));
static ADDRESS *setctluser __P((char *, int, ENVELOPE *));
#if _FFR_RHS
static int sm_strshufflecmp __P((char *, char *));
static void init_shuffle_alphabet __P(());
#endif /* _FFR_RHS */
static int workcmpf0();
static int workcmpf1();
static int workcmpf2();
static int workcmpf3();
static int workcmpf4();
static int randi = 3; /* index for workcmpf5() */
static int workcmpf5();
static int workcmpf6();
#if _FFR_RHS
static int workcmpf7();
#endif /* _FFR_RHS */
#if RANDOMSHIFT
# define get_rand_mod(m) ((get_random() >> RANDOMSHIFT) % (m))
#else /* RANDOMSHIFT */
# define get_rand_mod(m) (get_random() % (m))
#endif /* RANDOMSHIFT */
/*
** File system definition.
** Used to keep track of how much free space is available
** on a file system in which one or more queue directories reside.
*/
typedef struct filesys_shared FILESYS;
struct filesys_shared
{
dev_t fs_dev; /* unique device id */
long fs_avail; /* number of free blocks available */
long fs_blksize; /* block size, in bytes */
};
/* probably kept in shared memory */
static FILESYS FileSys[MAXFILESYS]; /* queue file systems */
static char *FSPath[MAXFILESYS]; /* pathnames for file systems */
#if SM_CONF_SHM
/*
** Shared memory data
**
** Current layout:
** size -- size of shared memory segment
** pid -- pid of owner, should be a unique id to avoid misinterpretations
** by other processes.
** tag -- should be a unique id to avoid misinterpretations by others.
** idea: hash over configuration data that will be stored here.
** NumFileSys -- number of file systems.
** FileSys -- (arrary of) structure for used file systems.
** RSATmpCnt -- counter for number of uses of ephemeral RSA key.
** QShm -- (array of) structure for information about queue directories.
*/
/*
** Queue data in shared memory
*/
typedef struct queue_shared QUEUE_SHM_T;
struct queue_shared
{
int qs_entries; /* number of entries */
/* XXX more to follow? */
};
static void *Pshm; /* pointer to shared memory */
static FILESYS *PtrFileSys; /* pointer to queue file system array */
int ShmId = SM_SHM_NO_ID; /* shared memory id */
static QUEUE_SHM_T *QShm; /* pointer to shared queue data */
static size_t shms;
# define SHM_OFF_PID(p) (((char *) (p)) + sizeof(int))
# define SHM_OFF_TAG(p) (((char *) (p)) + sizeof(pid_t) + sizeof(int))
# define SHM_OFF_HEAD (sizeof(pid_t) + sizeof(int) * 2)
/* how to access FileSys */
# define FILE_SYS(i) (PtrFileSys[i])
/* first entry is a tag, for now just the size */
# define OFF_FILE_SYS(p) (((char *) (p)) + SHM_OFF_HEAD)
/* offset for PNumFileSys */
# define OFF_NUM_FILE_SYS(p) (((char *) (p)) + SHM_OFF_HEAD + sizeof(FileSys))
/* offset for PRSATmpCnt */
# define OFF_RSA_TMP_CNT(p) (((char *) (p)) + SHM_OFF_HEAD + sizeof(FileSys) + sizeof(int))
int *PRSATmpCnt;
/* offset for queue_shm */
# define OFF_QUEUE_SHM(p) (((char *) (p)) + SHM_OFF_HEAD + sizeof(FileSys) + sizeof(int) * 2)
# define QSHM_ENTRIES(i) QShm[i].qs_entries
/* basic size of shared memory segment */
# define SM_T_SIZE (SHM_OFF_HEAD + sizeof(FileSys) + sizeof(int) * 2)
static unsigned int hash_q __P((char *, unsigned int));
/*
** HASH_Q -- simple hash function
**
** Parameters:
** p -- string to hash.
** h -- hash start value (from previous run).
**
** Returns:
** hash value.
*/
static unsigned int
hash_q(p, h)
char *p;
unsigned int h;
{
int c, d;
while (*p != '\0')
{
d = *p++;
c = d;
c ^= c<<6;
h += (c<<11) ^ (c>>1);
h ^= (d<<14) + (d<<7) + (d<<4) + d;
}
return h;
}
#else /* SM_CONF_SHM */
# define FILE_SYS(i) FileSys[i]
#endif /* SM_CONF_SHM */
/* access to the various components of file system data */
#define FILE_SYS_NAME(i) FSPath[i]
#define FILE_SYS_AVAIL(i) FILE_SYS(i).fs_avail
#define FILE_SYS_BLKSIZE(i) FILE_SYS(i).fs_blksize
#define FILE_SYS_DEV(i) FILE_SYS(i).fs_dev
/*
** Current qf file field assignments:
**
** A AUTH= parameter
** B body type
** C controlling user
** D data file name
** d data file directory name (added in 8.12)
** E error recipient
** F flag bits
** G free (was: queue delay algorithm if _FFR_QUEUEDELAY)
** H header
** I data file's inode number
** K time of last delivery attempt
** L Solaris Content-Length: header (obsolete)
** M message
** N number of delivery attempts
** P message priority
** q quarantine reason
** Q original recipient (ORCPT=)
** r final recipient (Final-Recipient: DSN field)
** R recipient
** S sender
** T init time
** V queue file version
** X free (was: character set if _FFR_SAVE_CHARSET)
** Y free (was: current delay if _FFR_QUEUEDELAY)
** Z original envelope id from ESMTP
** ! deliver by (added in 8.12)
** $ define macro
** . terminate file
*/
/*
** QUEUEUP -- queue a message up for future transmission.
**
** Parameters:
** e -- the envelope to queue up.
** announce -- if true, tell when you are queueing up.
** msync -- if true, then fsync() if SuperSafe interactive mode.
**
** Returns:
** none.
**
** Side Effects:
** The current request is saved in a control file.
** The queue file is left locked.
*/
void
queueup(e, announce, msync)
register ENVELOPE *e;
bool announce;
bool msync;
{
register SM_FILE_T *tfp;
register HDR *h;
register ADDRESS *q;
int tfd = -1;
int i;
bool newid;
register char *p;
MAILER nullmailer;
MCI mcibuf;
char qf[MAXPATHLEN];
char tf[MAXPATHLEN];
char df[MAXPATHLEN];
char buf[MAXLINE];
/*
** Create control file.
*/
#define OPEN_TF do \
{ \
MODE_T oldumask = 0; \
\
if (bitset(S_IWGRP, QueueFileMode)) \
oldumask = umask(002); \
tfd = open(tf, TF_OPEN_FLAGS, QueueFileMode); \
if (bitset(S_IWGRP, QueueFileMode)) \
(void) umask(oldumask); \
} while (0)
newid = (e->e_id == NULL) || !bitset(EF_INQUEUE, e->e_flags);
(void) sm_strlcpy(tf, queuename(e, NEWQFL_LETTER), sizeof tf);
tfp = e->e_lockfp;
if (tfp == NULL && newid)
{
/*
** open qf file directly: this will give an error if the file
** already exists and hence prevent problems if a queue-id
** is reused (e.g., because the clock is set back).
*/
(void) sm_strlcpy(tf, queuename(e, ANYQFL_LETTER), sizeof tf);
OPEN_TF;
if (tfd < 0 ||
#if !SM_OPEN_EXLOCK
!lockfile(tfd, tf, NULL, LOCK_EX|LOCK_NB) ||
#endif /* !SM_OPEN_EXLOCK */
(tfp = sm_io_open(SmFtStdiofd, SM_TIME_DEFAULT,
(void *) &tfd, SM_IO_WRONLY,
NULL)) == NULL)
{
int save_errno = errno;
printopenfds(true);
errno = save_errno;
syserr("!queueup: cannot create queue file %s, euid=%d, fd=%d, fp=%p",
tf, (int) geteuid(), tfd, tfp);
/* NOTREACHED */
}
e->e_lockfp = tfp;
upd_qs(e, 1, 0, "queueup");
}
/* if newid, write the queue file directly (instead of temp file) */
if (!newid)
{
/* get a locked tf file */
for (i = 0; i < 128; i++)
{
if (tfd < 0)
{
OPEN_TF;
if (tfd < 0)
{
if (errno != EEXIST)
break;
if (LogLevel > 0 && (i % 32) == 0)
sm_syslog(LOG_ALERT, e->e_id,
"queueup: cannot create %s, uid=%d: %s",
tf, (int) geteuid(),
sm_errstring(errno));
}
#if SM_OPEN_EXLOCK
else
break;
#endif /* SM_OPEN_EXLOCK */
}
if (tfd >= 0)
{
#if SM_OPEN_EXLOCK
/* file is locked by open() */
break;
#else /* SM_OPEN_EXLOCK */
if (lockfile(tfd, tf, NULL, LOCK_EX|LOCK_NB))
break;
else
#endif /* SM_OPEN_EXLOCK */
if (LogLevel > 0 && (i % 32) == 0)
sm_syslog(LOG_ALERT, e->e_id,
"queueup: cannot lock %s: %s",
tf, sm_errstring(errno));
if ((i % 32) == 31)
{
(void) close(tfd);
tfd = -1;
}
}
if ((i % 32) == 31)
{
/* save the old temp file away */
(void) rename(tf, queuename(e, TEMPQF_LETTER));
}
else
(void) sleep(i % 32);
}
if (tfd < 0 || (tfp = sm_io_open(SmFtStdiofd, SM_TIME_DEFAULT,
(void *) &tfd, SM_IO_WRONLY_B,
NULL)) == NULL)
{
int save_errno = errno;
printopenfds(true);
errno = save_errno;
syserr("!queueup: cannot create queue temp file %s, uid=%d",
tf, (int) geteuid());
}
}
if (tTd(40, 1))
sm_dprintf("\n>>>>> queueing %s/%s%s >>>>>\n",
qid_printqueue(e->e_qgrp, e->e_qdir),
queuename(e, ANYQFL_LETTER),
newid ? " (new id)" : "");
if (tTd(40, 3))
{
sm_dprintf(" e_flags=");
printenvflags(e);
}
if (tTd(40, 32))
{
sm_dprintf(" sendq=");
printaddr(sm_debug_file(), e->e_sendqueue, true);
}
if (tTd(40, 9))
{
sm_dprintf(" tfp=");
dumpfd(sm_io_getinfo(tfp, SM_IO_WHAT_FD, NULL), true, false);
sm_dprintf(" lockfp=");
if (e->e_lockfp == NULL)
sm_dprintf("NULL\n");
else
dumpfd(sm_io_getinfo(e->e_lockfp, SM_IO_WHAT_FD, NULL),
true, false);
}
/*
** If there is no data file yet, create one.
*/
(void) sm_strlcpy(df, queuename(e, DATAFL_LETTER), sizeof df);
if (bitset(EF_HAS_DF, e->e_flags))
{
if (e->e_dfp != NULL &&
SuperSafe != SAFE_REALLY &&
SuperSafe != SAFE_REALLY_POSTMILTER &&
sm_io_setinfo(e->e_dfp, SM_BF_COMMIT, NULL) < 0 &&
errno != EINVAL)
{
syserr("!queueup: cannot commit data file %s, uid=%d",
queuename(e, DATAFL_LETTER), (int) geteuid());
}
if (e->e_dfp != NULL &&
SuperSafe == SAFE_INTERACTIVE && msync)
{
if (tTd(40,32))
sm_syslog(LOG_INFO, e->e_id,
"queueup: fsync(e->e_dfp)");
if (fsync(sm_io_getinfo(e->e_dfp, SM_IO_WHAT_FD,
NULL)) < 0)
{
if (newid)
syserr("!552 Error writing data file %s",
df);
else
syserr("!452 Error writing data file %s",
df);
}
}
}
else
{
int dfd;
MODE_T oldumask = 0;
register SM_FILE_T *dfp = NULL;
struct stat stbuf;
if (e->e_dfp != NULL &&
sm_io_getinfo(e->e_dfp, SM_IO_WHAT_ISTYPE, BF_FILE_TYPE))
syserr("committing over bf file");
if (bitset(S_IWGRP, QueueFileMode))
oldumask = umask(002);
dfd = open(df, O_WRONLY|O_CREAT|O_TRUNC|QF_O_EXTRA,
QueueFileMode);
if (bitset(S_IWGRP, QueueFileMode))
(void) umask(oldumask);
if (dfd < 0 || (dfp = sm_io_open(SmFtStdiofd, SM_TIME_DEFAULT,
(void *) &dfd, SM_IO_WRONLY_B,
NULL)) == NULL)
syserr("!queueup: cannot create data temp file %s, uid=%d",
df, (int) geteuid());
if (fstat(dfd, &stbuf) < 0)
e->e_dfino = -1;
else
{
e->e_dfdev = stbuf.st_dev;
e->e_dfino = ST_INODE(stbuf);
}
e->e_flags |= EF_HAS_DF;
memset(&mcibuf, '\0', sizeof mcibuf);
mcibuf.mci_out = dfp;
mcibuf.mci_mailer = FileMailer;
(*e->e_putbody)(&mcibuf, e, NULL);
if (SuperSafe == SAFE_REALLY ||
SuperSafe == SAFE_REALLY_POSTMILTER ||
(SuperSafe == SAFE_INTERACTIVE && msync))
{
if (tTd(40,32))
sm_syslog(LOG_INFO, e->e_id,
"queueup: fsync(dfp)");
if (fsync(sm_io_getinfo(dfp, SM_IO_WHAT_FD, NULL)) < 0)
{
if (newid)
syserr("!552 Error writing data file %s",
df);
else
syserr("!452 Error writing data file %s",
df);
}
}
if (sm_io_close(dfp, SM_TIME_DEFAULT) < 0)
syserr("!queueup: cannot save data temp file %s, uid=%d",
df, (int) geteuid());
e->e_putbody = putbody;
}
/*
** Output future work requests.
** Priority and creation time should be first, since
** they are required by gatherq.
*/
/* output queue version number (must be first!) */
(void) sm_io_fprintf(tfp, SM_TIME_DEFAULT, "V%d\n", QF_VERSION);
/* output creation time */
(void) sm_io_fprintf(tfp, SM_TIME_DEFAULT, "T%ld\n", (long) e->e_ctime);
/* output last delivery time */
(void) sm_io_fprintf(tfp, SM_TIME_DEFAULT, "K%ld\n", (long) e->e_dtime);
/* output number of delivery attempts */
(void) sm_io_fprintf(tfp, SM_TIME_DEFAULT, "N%d\n", e->e_ntries);
/* output message priority */
(void) sm_io_fprintf(tfp, SM_TIME_DEFAULT, "P%ld\n", e->e_msgpriority);
/*
** If data file is in a different directory than the queue file,
** output a "d" record naming the directory of the data file.
*/
if (e->e_dfqgrp != e->e_qgrp)
{
(void) sm_io_fprintf(tfp, SM_TIME_DEFAULT, "d%s\n",
Queue[e->e_dfqgrp]->qg_qpaths[e->e_dfqdir].qp_name);
}
/* output inode number of data file */
/* XXX should probably include device major/minor too */
if (e->e_dfino != -1)
{
(void) sm_io_fprintf(tfp, SM_TIME_DEFAULT, "I%ld/%ld/%llu\n",
(long) major(e->e_dfdev),
(long) minor(e->e_dfdev),
(ULONGLONG_T) e->e_dfino);
}
/* output body type */
if (e->e_bodytype != NULL)
(void) sm_io_fprintf(tfp, SM_TIME_DEFAULT, "B%s\n",
denlstring(e->e_bodytype, true, false));
/* quarantine reason */
if (e->e_quarmsg != NULL)
(void) sm_io_fprintf(tfp, SM_TIME_DEFAULT, "q%s\n",
denlstring(e->e_quarmsg, true, false));
/* message from envelope, if it exists */
if (e->e_message != NULL)
(void) sm_io_fprintf(tfp, SM_TIME_DEFAULT, "M%s\n",
denlstring(e->e_message, true, false));
/* send various flag bits through */
p = buf;
if (bitset(EF_WARNING, e->e_flags))
*p++ = 'w';
if (bitset(EF_RESPONSE, e->e_flags))
*p++ = 'r';
if (bitset(EF_HAS8BIT, e->e_flags))
*p++ = '8';
if (bitset(EF_DELETE_BCC, e->e_flags))
*p++ = 'b';
if (bitset(EF_RET_PARAM, e->e_flags))
*p++ = 'd';
if (bitset(EF_NO_BODY_RETN, e->e_flags))
*p++ = 'n';
if (bitset(EF_SPLIT, e->e_flags))
*p++ = 's';
*p++ = '\0';
if (buf[0] != '\0')
(void) sm_io_fprintf(tfp, SM_TIME_DEFAULT, "F%s\n", buf);
/* save $={persistentMacros} macro values */
queueup_macros(macid("{persistentMacros}"), tfp, e);
/* output name of sender */
if (bitnset(M_UDBENVELOPE, e->e_from.q_mailer->m_flags))
p = e->e_sender;
else
p = e->e_from.q_paddr;
(void) sm_io_fprintf(tfp, SM_TIME_DEFAULT, "S%s\n",
denlstring(p, true, false));
/* output ESMTP-supplied "original" information */
if (e->e_envid != NULL)
(void) sm_io_fprintf(tfp, SM_TIME_DEFAULT, "Z%s\n",
denlstring(e->e_envid, true, false));
/* output AUTH= parameter */
if (e->e_auth_param != NULL)
(void) sm_io_fprintf(tfp, SM_TIME_DEFAULT, "A%s\n",
denlstring(e->e_auth_param, true, false));
if (e->e_dlvr_flag != 0)
(void) sm_io_fprintf(tfp, SM_TIME_DEFAULT, "!%c %ld\n",
(char) e->e_dlvr_flag, e->e_deliver_by);
/* output list of recipient addresses */
printctladdr(NULL, NULL);
for (q = e->e_sendqueue; q != NULL; q = q->q_next)
{
if (!QS_IS_UNDELIVERED(q->q_state))
continue;
/* message for this recipient, if it exists */
if (q->q_message != NULL)
(void) sm_io_fprintf(tfp, SM_TIME_DEFAULT, "M%s\n",
denlstring(q->q_message, true,
false));
printctladdr(q, tfp);
if (q->q_orcpt != NULL)
(void) sm_io_fprintf(tfp, SM_TIME_DEFAULT, "Q%s\n",
denlstring(q->q_orcpt, true,
false));
if (q->q_finalrcpt != NULL)
(void) sm_io_fprintf(tfp, SM_TIME_DEFAULT, "r%s\n",
denlstring(q->q_finalrcpt, true,
false));
(void) sm_io_putc(tfp, SM_TIME_DEFAULT, 'R');
if (bitset(QPRIMARY, q->q_flags))
(void) sm_io_putc(tfp, SM_TIME_DEFAULT, 'P');
if (bitset(QHASNOTIFY, q->q_flags))
(void) sm_io_putc(tfp, SM_TIME_DEFAULT, 'N');
if (bitset(QPINGONSUCCESS, q->q_flags))
(void) sm_io_putc(tfp, SM_TIME_DEFAULT, 'S');
if (bitset(QPINGONFAILURE, q->q_flags))
(void) sm_io_putc(tfp, SM_TIME_DEFAULT, 'F');
if (bitset(QPINGONDELAY, q->q_flags))
(void) sm_io_putc(tfp, SM_TIME_DEFAULT, 'D');
if (q->q_alias != NULL &&
bitset(QALIAS, q->q_alias->q_flags))
(void) sm_io_putc(tfp, SM_TIME_DEFAULT, 'A');
(void) sm_io_putc(tfp, SM_TIME_DEFAULT, ':');
(void) sm_io_fprintf(tfp, SM_TIME_DEFAULT, "%s\n",
denlstring(q->q_paddr, true, false));
if (announce)
{
char *tag = "queued";
if (e->e_quarmsg != NULL)
tag = "quarantined";
e->e_to = q->q_paddr;
message(tag);
if (LogLevel > 8)
logdelivery(q->q_mailer, NULL, q->q_status,
tag, NULL, (time_t) 0, e);
e->e_to = NULL;
}
if (tTd(40, 1))
{
sm_dprintf("queueing ");
printaddr(sm_debug_file(), q, false);
}
}
/*
** Output headers for this message.
** Expand macros completely here. Queue run will deal with
** everything as absolute headers.
** All headers that must be relative to the recipient
** can be cracked later.
** We set up a "null mailer" -- i.e., a mailer that will have
** no effect on the addresses as they are output.
*/
memset((char *) &nullmailer, '\0', sizeof nullmailer);
nullmailer.m_re_rwset = nullmailer.m_rh_rwset =
nullmailer.m_se_rwset = nullmailer.m_sh_rwset = -1;
nullmailer.m_eol = "\n";
memset(&mcibuf, '\0', sizeof mcibuf);
mcibuf.mci_mailer = &nullmailer;
mcibuf.mci_out = tfp;
macdefine(&e->e_macro, A_PERM, 'g', "\201f");
for (h = e->e_header; h != NULL; h = h->h_link)
{
if (h->h_value == NULL)
continue;
/* don't output resent headers on non-resent messages */
if (bitset(H_RESENT, h->h_flags) &&
!bitset(EF_RESENT, e->e_flags))
continue;
/* expand macros; if null, don't output header at all */
if (bitset(H_DEFAULT, h->h_flags))
{
(void) expand(h->h_value, buf, sizeof buf, e);
if (buf[0] == '\0')
continue;
}
/* output this header */
(void) sm_io_fprintf(tfp, SM_TIME_DEFAULT, "H?");
/* output conditional macro if present */
if (h->h_macro != '\0')
{
if (bitset(0200, h->h_macro))
(void) sm_io_fprintf(tfp, SM_TIME_DEFAULT,
"${%s}",
macname(bitidx(h->h_macro)));
else
(void) sm_io_fprintf(tfp, SM_TIME_DEFAULT,
"$%c", h->h_macro);
}
else if (!bitzerop(h->h_mflags) &&
bitset(H_CHECK|H_ACHECK, h->h_flags))
{
int j;
/* if conditional, output the set of conditions */
for (j = '\0'; j <= '\177'; j++)
if (bitnset(j, h->h_mflags))
(void) sm_io_putc(tfp, SM_TIME_DEFAULT,
j);
}
(void) sm_io_putc(tfp, SM_TIME_DEFAULT, '?');
/* output the header: expand macros, convert addresses */
if (bitset(H_DEFAULT, h->h_flags) &&
!bitset(H_BINDLATE, h->h_flags))
{
(void) sm_io_fprintf(tfp, SM_TIME_DEFAULT, "%s: %s\n",
h->h_field,
denlstring(buf, false, true));
}
else if (bitset(H_FROM|H_RCPT, h->h_flags) &&
!bitset(H_BINDLATE, h->h_flags))
{
bool oldstyle = bitset(EF_OLDSTYLE, e->e_flags);
SM_FILE_T *savetrace = TrafficLogFile;
TrafficLogFile = NULL;
if (bitset(H_FROM, h->h_flags))
oldstyle = false;
commaize(h, h->h_value, oldstyle, &mcibuf, e);
TrafficLogFile = savetrace;
}
else
{
(void) sm_io_fprintf(tfp, SM_TIME_DEFAULT, "%s: %s\n",
h->h_field,
denlstring(h->h_value, false,
true));
}
}
/*
** Clean up.
**
** Write a terminator record -- this is to prevent
** scurrilous crackers from appending any data.
*/
(void) sm_io_fprintf(tfp, SM_TIME_DEFAULT, ".\n");
if (sm_io_flush(tfp, SM_TIME_DEFAULT) != 0 ||
((SuperSafe == SAFE_REALLY ||
SuperSafe == SAFE_REALLY_POSTMILTER ||
(SuperSafe == SAFE_INTERACTIVE && msync)) &&
fsync(sm_io_getinfo(tfp, SM_IO_WHAT_FD, NULL)) < 0) ||
sm_io_error(tfp))
{
if (newid)
syserr("!552 Error writing control file %s", tf);
else
syserr("!452 Error writing control file %s", tf);
}
if (!newid)
{
char new = queue_letter(e, ANYQFL_LETTER);
/* rename (locked) tf to be (locked) [qh]f */
(void) sm_strlcpy(qf, queuename(e, ANYQFL_LETTER),
sizeof qf);
if (rename(tf, qf) < 0)
syserr("cannot rename(%s, %s), uid=%d",
tf, qf, (int) geteuid());
else
{
/*
** Check if type has changed and only
** remove the old item if the rename above
** succeeded.
*/
if (e->e_qfletter != '\0' &&
e->e_qfletter != new)
{
if (tTd(40, 5))
{
sm_dprintf("type changed from %c to %c\n",
e->e_qfletter, new);
}
if (unlink(queuename(e, e->e_qfletter)) < 0)
{
/* XXX: something more drastic? */
if (LogLevel > 0)
sm_syslog(LOG_ERR, e->e_id,
"queueup: unlink(%s) failed: %s",
queuename(e, e->e_qfletter),
sm_errstring(errno));
}
}
}
e->e_qfletter = new;
/*
** fsync() after renaming to make sure metadata is
** written to disk on filesystems in which renames are
** not guaranteed.
*/
if (SuperSafe != SAFE_NO)
{
/* for softupdates */
if (tfd >= 0 && fsync(tfd) < 0)
{
syserr("!queueup: cannot fsync queue temp file %s",
tf);
}
SYNC_DIR(qf, true);
}
/* close and unlock old (locked) queue file */
if (e->e_lockfp != NULL)
(void) sm_io_close(e->e_lockfp, SM_TIME_DEFAULT);
e->e_lockfp = tfp;
/* save log info */
if (LogLevel > 79)
sm_syslog(LOG_DEBUG, e->e_id, "queueup %s", qf);
}
else
{
/* save log info */
if (LogLevel > 79)
sm_syslog(LOG_DEBUG, e->e_id, "queueup %s", tf);
e->e_qfletter = queue_letter(e, ANYQFL_LETTER);
}
errno = 0;
e->e_flags |= EF_INQUEUE;
if (tTd(40, 1))
sm_dprintf("<<<<< done queueing %s <<<<<\n\n", e->e_id);
return;
}
/*
** PRINTCTLADDR -- print control address to file.
**
** Parameters:
** a -- address.
** tfp -- file pointer.
**
** Returns:
** none.
**
** Side Effects:
** The control address (if changed) is printed to the file.
** The last control address and uid are saved.
*/
static void
printctladdr(a, tfp)
register ADDRESS *a;
SM_FILE_T *tfp;
{
char *user;
register ADDRESS *q;
uid_t uid;
gid_t gid;
static ADDRESS *lastctladdr = NULL;
static uid_t lastuid;
/* initialization */
if (a == NULL || a->q_alias == NULL || tfp == NULL)
{
if (lastctladdr != NULL && tfp != NULL)
(void) sm_io_fprintf(tfp, SM_TIME_DEFAULT, "C\n");
lastctladdr = NULL;
lastuid = 0;
return;
}
/* find the active uid */
q = getctladdr(a);
if (q == NULL)
{
user = NULL;
uid = 0;
gid = 0;
}
else
{
user = q->q_ruser != NULL ? q->q_ruser : q->q_user;
uid = q->q_uid;
gid = q->q_gid;
}
a = a->q_alias;
/* check to see if this is the same as last time */
if (lastctladdr != NULL && uid == lastuid &&
strcmp(lastctladdr->q_paddr, a->q_paddr) == 0)
return;
lastuid = uid;
lastctladdr = a;
if (uid == 0 || user == NULL || user[0] == '\0')
(void) sm_io_fprintf(tfp, SM_TIME_DEFAULT, "C");
else
(void) sm_io_fprintf(tfp, SM_TIME_DEFAULT, "C%s:%ld:%ld",
denlstring(user, true, false), (long) uid,
(long) gid);
(void) sm_io_fprintf(tfp, SM_TIME_DEFAULT, ":%s\n",
denlstring(a->q_paddr, true, false));
}
/*
** RUNNERS_SIGTERM -- propagate a SIGTERM to queue runner process
**
** This propagates the signal to the child processes that are queue
** runners. This is for a queue runner "cleanup". After all of the
** child queue runner processes are signaled (it should be SIGTERM
** being the sig) then the old signal handler (Oldsh) is called
** to handle any cleanup set for this process (provided it is not
** SIG_DFL or SIG_IGN). The signal may not be handled immediately
** if the BlockOldsh flag is set. If the current process doesn't
** have a parent then handle the signal immediately, regardless of
** BlockOldsh.
**
** Parameters:
** sig -- the signal number being sent
**
** Returns:
** none.
**
** Side Effects:
** Sets the NoMoreRunners boolean to true to stop more runners
** from being started in runqueue().
**
** NOTE: THIS CAN BE CALLED FROM A SIGNAL HANDLER. DO NOT ADD
** ANYTHING TO THIS ROUTINE UNLESS YOU KNOW WHAT YOU ARE
** DOING.
*/
static bool volatile NoMoreRunners = false;
static sigfunc_t Oldsh_term = SIG_DFL;
static sigfunc_t Oldsh_hup = SIG_DFL;
static sigfunc_t volatile Oldsh = SIG_DFL;
static bool BlockOldsh = false;
static int volatile Oldsig = 0;
static SIGFUNC_DECL runners_sigterm __P((int));
static SIGFUNC_DECL runners_sighup __P((int));
static SIGFUNC_DECL
runners_sigterm(sig)
int sig;
{
int save_errno = errno;
FIX_SYSV_SIGNAL(sig, runners_sigterm);
errno = save_errno;
CHECK_CRITICAL(sig);
NoMoreRunners = true;
Oldsh = Oldsh_term;
Oldsig = sig;
proc_list_signal(PROC_QUEUE, sig);
if (!BlockOldsh || getppid() <= 1)
{
/* Check that a valid 'old signal handler' is callable */
if (Oldsh_term != SIG_DFL && Oldsh_term != SIG_IGN &&
Oldsh_term != runners_sigterm)
(*Oldsh_term)(sig);
}
errno = save_errno;
return SIGFUNC_RETURN;
}
/*
** RUNNERS_SIGHUP -- propagate a SIGHUP to queue runner process
**
** This propagates the signal to the child processes that are queue
** runners. This is for a queue runner "cleanup". After all of the
** child queue runner processes are signaled (it should be SIGHUP
** being the sig) then the old signal handler (Oldsh) is called to
** handle any cleanup set for this process (provided it is not SIG_DFL
** or SIG_IGN). The signal may not be handled immediately if the
** BlockOldsh flag is set. If the current process doesn't have
** a parent then handle the signal immediately, regardless of
** BlockOldsh.
**
** Parameters:
** sig -- the signal number being sent
**
** Returns:
** none.
**
** Side Effects:
** Sets the NoMoreRunners boolean to true to stop more runners
** from being started in runqueue().
**
** NOTE: THIS CAN BE CALLED FROM A SIGNAL HANDLER. DO NOT ADD
** ANYTHING TO THIS ROUTINE UNLESS YOU KNOW WHAT YOU ARE
** DOING.
*/
static SIGFUNC_DECL
runners_sighup(sig)
int sig;
{
int save_errno = errno;
FIX_SYSV_SIGNAL(sig, runners_sighup);
errno = save_errno;
CHECK_CRITICAL(sig);
NoMoreRunners = true;
Oldsh = Oldsh_hup;
Oldsig = sig;
proc_list_signal(PROC_QUEUE, sig);
if (!BlockOldsh || getppid() <= 1)
{
/* Check that a valid 'old signal handler' is callable */
if (Oldsh_hup != SIG_DFL && Oldsh_hup != SIG_IGN &&
Oldsh_hup != runners_sighup)
(*Oldsh_hup)(sig);
}
errno = save_errno;
return SIGFUNC_RETURN;
}
/*
** MARK_WORK_GROUP_RESTART -- mark a work group as needing a restart
**
** Sets a workgroup for restarting.
**
** Parameters:
** wgrp -- the work group id to restart.
** reason -- why (signal?), -1 to turn off restart
**
** Returns:
** none.
**
** Side effects:
** May set global RestartWorkGroup to true.
**
** NOTE: THIS CAN BE CALLED FROM A SIGNAL HANDLER. DO NOT ADD
** ANYTHING TO THIS ROUTINE UNLESS YOU KNOW WHAT YOU ARE
** DOING.
*/
void
mark_work_group_restart(wgrp, reason)
int wgrp;
int reason;
{
if (wgrp < 0 || wgrp > NumWorkGroups)
return;
WorkGrp[wgrp].wg_restart = reason;
if (reason >= 0)
RestartWorkGroup = true;
}
/*
** RESTART_MARKED_WORK_GROUPS -- restart work groups marked as needing restart
**
** Restart any workgroup marked as needing a restart provided more
** runners are allowed.
**
** Parameters:
** none.
**
** Returns:
** none.
**
** Side effects:
** Sets global RestartWorkGroup to false.
*/
void
restart_marked_work_groups()
{
int i;
int wasblocked;
if (NoMoreRunners)
return;
/* Block SIGCHLD so reapchild() doesn't mess with us */
wasblocked = sm_blocksignal(SIGCHLD);
for (i = 0; i < NumWorkGroups; i++)
{
if (WorkGrp[i].wg_restart >= 0)
{
if (LogLevel > 8)
sm_syslog(LOG_ERR, NOQID,
"restart queue runner=%d due to signal 0x%x",
i, WorkGrp[i].wg_restart);
restart_work_group(i);
}
}
RestartWorkGroup = false;
if (wasblocked == 0)
(void) sm_releasesignal(SIGCHLD);
}
/*
** RESTART_WORK_GROUP -- restart a specific work group
**
** Restart a specific workgroup provided more runners are allowed.
** If the requested work group has been restarted too many times log
** this and refuse to restart.
**
** Parameters:
** wgrp -- the work group id to restart
**
** Returns:
** none.
**
** Side Effects:
** starts another process doing the work of wgrp
*/
#define MAX_PERSIST_RESTART 10 /* max allowed number of restarts */
static void
restart_work_group(wgrp)
int wgrp;
{
if (NoMoreRunners ||
wgrp < 0 || wgrp > NumWorkGroups)
return;
WorkGrp[wgrp].wg_restart = -1;
if (WorkGrp[wgrp].wg_restartcnt < MAX_PERSIST_RESTART)
{
/* avoid overflow; increment here */
WorkGrp[wgrp].wg_restartcnt++;
(void) run_work_group(wgrp, RWG_FORK|RWG_PERSISTENT|RWG_RUNALL);
}
else
{
sm_syslog(LOG_ERR, NOQID,
"ERROR: persistent queue runner=%d restarted too many times, queue runner lost",
wgrp);
}
}
/*
** SCHEDULE_QUEUE_RUNS -- schedule the next queue run for a work group.
**
** Parameters:
** runall -- schedule even if individual bit is not set.
** wgrp -- the work group id to schedule.
** didit -- the queue run was performed for this work group.
**
** Returns:
** nothing
*/
#define INCR_MOD(v, m) if (++v >= m) \
v = 0; \
else
static void
schedule_queue_runs(runall, wgrp, didit)
bool runall;
int wgrp;
bool didit;
{
int qgrp, cgrp, endgrp;
#if _FFR_QUEUE_SCHED_DBG
time_t lastsched;
bool sched;
#endif /* _FFR_QUEUE_SCHED_DBG */
time_t now;
time_t minqintvl;
/*
** This is a bit ugly since we have to duplicate the
** code that "walks" through a work queue group.
*/
now = curtime();
minqintvl = 0;
cgrp = endgrp = WorkGrp[wgrp].wg_curqgrp;
do
{
time_t qintvl;
#if _FFR_QUEUE_SCHED_DBG
lastsched = 0;
sched = false;
#endif /* _FFR_QUEUE_SCHED_DBG */
qgrp = WorkGrp[wgrp].wg_qgs[cgrp]->qg_index;
if (Queue[qgrp]->qg_queueintvl > 0)
qintvl = Queue[qgrp]->qg_queueintvl;
else if (QueueIntvl > 0)
qintvl = QueueIntvl;
else
qintvl = (time_t) 0;
#if _FFR_QUEUE_SCHED_DBG
lastsched = Queue[qgrp]->qg_nextrun;
#endif /* _FFR_QUEUE_SCHED_DBG */
if ((runall || Queue[qgrp]->qg_nextrun <= now) && qintvl > 0)
{
#if _FFR_QUEUE_SCHED_DBG
sched = true;
#endif /* _FFR_QUEUE_SCHED_DBG */
if (minqintvl == 0 || qintvl < minqintvl)
minqintvl = qintvl;
/*
** Only set a new time if a queue run was performed
** for this queue group. If the queue was not run,
** we could starve it by setting a new time on each
** call.
*/
if (didit)
Queue[qgrp]->qg_nextrun += qintvl;
}
#if _FFR_QUEUE_SCHED_DBG
if (tTd(69, 10))
sm_syslog(LOG_INFO, NOQID,
"sqr: wgrp=%d, cgrp=%d, qgrp=%d, intvl=%ld, QI=%ld, runall=%d, lastrun=%ld, nextrun=%ld, sched=%d",
wgrp, cgrp, qgrp, Queue[qgrp]->qg_queueintvl,
QueueIntvl, runall, lastsched,
Queue[qgrp]->qg_nextrun, sched);
#endif /* _FFR_QUEUE_SCHED_DBG */
INCR_MOD(cgrp, WorkGrp[wgrp].wg_numqgrp);
} while (endgrp != cgrp);
if (minqintvl > 0)
(void) sm_setevent(minqintvl, runqueueevent, 0);
}
#if _FFR_QUEUE_RUN_PARANOIA
/*
** CHECKQUEUERUNNER -- check whether a queue group hasn't been run.
**
** Use this if events may get lost and hence queue runners may not
** be started and mail will pile up in a queue.
**
** Parameters:
** none.
**
** Returns:
** true if a queue run is necessary.
**
** Side Effects:
** may schedule a queue run.
*/
bool
checkqueuerunner()
{
int qgrp;
time_t now, minqintvl;
now = curtime();
minqintvl = 0;
for (qgrp = 0; qgrp < NumQueue && Queue[qgrp] != NULL; qgrp++)
{
time_t qintvl;
if (Queue[qgrp]->qg_queueintvl > 0)
qintvl = Queue[qgrp]->qg_queueintvl;
else if (QueueIntvl > 0)
qintvl = QueueIntvl;
else
qintvl = (time_t) 0;
if (Queue[qgrp]->qg_nextrun <= now - qintvl)
{
if (minqintvl == 0 || qintvl < minqintvl)
minqintvl = qintvl;
if (LogLevel > 1)
sm_syslog(LOG_WARNING, NOQID,
"checkqueuerunner: queue %d should have been run at %s, queue interval %ld",
qgrp,
arpadate(ctime(&Queue[qgrp]->qg_nextrun)),
qintvl);
}
}
if (minqintvl > 0)
{
(void) sm_setevent(minqintvl, runqueueevent, 0);
return true;
}
return false;
}
#endif /* _FFR_QUEUE_RUN_PARANOIA */
/*
** RUNQUEUE -- run the jobs in the queue.
**
** Gets the stuff out of the queue in some presumably logical
** order and processes them.
**
** Parameters:
** forkflag -- true if the queue scanning should be done in
** a child process. We double-fork so it is not our
** child and we don't have to clean up after it.
** false can be ignored if we have multiple queues.
** verbose -- if true, print out status information.
** persistent -- persistent queue runner?
** runall -- run all groups or only a subset (DoQueueRun)?
**
** Returns:
** true if the queue run successfully began.
**
** Side Effects:
** runs things in the mail queue using run_work_group().
** maybe schedules next queue run.
*/
static ENVELOPE QueueEnvelope; /* the queue run envelope */
static time_t LastQueueTime = 0; /* last time a queue ID assigned */
static pid_t LastQueuePid = -1; /* last PID which had a queue ID */
/* values for qp_supdirs */
#define QP_NOSUB 0x0000 /* No subdirectories */
#define QP_SUBDF 0x0001 /* "df" subdirectory */
#define QP_SUBQF 0x0002 /* "qf" subdirectory */
#define QP_SUBXF 0x0004 /* "xf" subdirectory */
bool
runqueue(forkflag, verbose, persistent, runall)
bool forkflag;
bool verbose;
bool persistent;
bool runall;
{
int i;
bool ret = true;
static int curnum = 0;
sigfunc_t cursh;
#if SM_HEAP_CHECK
SM_NONVOLATILE int oldgroup = 0;
if (sm_debug_active(&DebugLeakQ, 1))
{
oldgroup = sm_heap_group();
sm_heap_newgroup();
sm_dprintf("runqueue() heap group #%d\n", sm_heap_group());
}
#endif /* SM_HEAP_CHECK */
/* queue run has been started, don't do any more this time */
DoQueueRun = false;
/* more than one queue or more than one directory per queue */
if (!forkflag && !verbose &&
(WorkGrp[0].wg_qgs[0]->qg_numqueues > 1 || NumWorkGroups > 1 ||
WorkGrp[0].wg_numqgrp > 1))
forkflag = true;
/*
** For controlling queue runners via signals sent to this process.
** Oldsh* will get called too by runners_sig* (if it is not SIG_IGN
** or SIG_DFL) to preserve cleanup behavior. Now that this process
** will have children (and perhaps grandchildren) this handler will
** be left in place. This is because this process, once it has
** finished spinning off queue runners, may go back to doing something
** else (like being a daemon). And we still want on a SIG{TERM,HUP} to
** clean up the child queue runners. Only install 'runners_sig*' once
** else we'll get stuck looping forever.
*/
cursh = sm_signal(SIGTERM, runners_sigterm);
if (cursh != runners_sigterm)
Oldsh_term = cursh;
cursh = sm_signal(SIGHUP, runners_sighup);
if (cursh != runners_sighup)
Oldsh_hup = cursh;
for (i = 0; i < NumWorkGroups && !NoMoreRunners; i++)
{
int rwgflags = RWG_NONE;
/*
** If MaxQueueChildren active then test whether the start
** of the next queue group's additional queue runners (maximum)
** will result in MaxQueueChildren being exceeded.
**
** Note: do not use continue; even though another workgroup
** may have fewer queue runners, this would be "unfair",
** i.e., this work group might "starve" then.
*/
#if _FFR_QUEUE_SCHED_DBG
if (tTd(69, 10))
sm_syslog(LOG_INFO, NOQID,
"rq: curnum=%d, MaxQueueChildren=%d, CurRunners=%d, WorkGrp[curnum].wg_maxact=%d",
curnum, MaxQueueChildren, CurRunners,
WorkGrp[curnum].wg_maxact);
#endif /* _FFR_QUEUE_SCHED_DBG */
if (MaxQueueChildren > 0 &&
CurRunners + WorkGrp[curnum].wg_maxact > MaxQueueChildren)
break;
/*
** Pick up where we left off (curnum), in case we
** used up all the children last time without finishing.
** This give a round-robin fairness to queue runs.
**
** Increment CurRunners before calling run_work_group()
** to avoid a "race condition" with proc_list_drop() which
** decrements CurRunners if the queue runners terminate.
** Notice: CurRunners is an upper limit, in some cases
** (too few jobs in the queue) this value is larger than
** the actual number of queue runners. The discrepancy can
** increase if some queue runners "hang" for a long time.
*/
CurRunners += WorkGrp[curnum].wg_maxact;
if (forkflag)
rwgflags |= RWG_FORK;
if (verbose)
rwgflags |= RWG_VERBOSE;
if (persistent)
rwgflags |= RWG_PERSISTENT;
if (runall)
rwgflags |= RWG_RUNALL;
ret = run_work_group(curnum, rwgflags);
/*
** Failure means a message was printed for ETRN
** and subsequent queues are likely to fail as well.
** Decrement CurRunners in that case because
** none have been started.
*/
if (!ret)
{
CurRunners -= WorkGrp[curnum].wg_maxact;
break;
}
if (!persistent)
schedule_queue_runs(runall, curnum, true);
INCR_MOD(curnum, NumWorkGroups);
}
/* schedule left over queue runs */
if (i < NumWorkGroups && !NoMoreRunners && !persistent)
{
int h;
for (h = curnum; i < NumWorkGroups; i++)
{
schedule_queue_runs(runall, h, false);
INCR_MOD(h, NumWorkGroups);
}
}
#if SM_HEAP_CHECK
if (sm_debug_active(&DebugLeakQ, 1))
sm_heap_setgroup(oldgroup);
#endif /* SM_HEAP_CHECK */
return ret;
}
#if _FFR_SKIP_DOMAINS
/*
** SKIP_DOMAINS -- Skip 'skip' number of domains in the WorkQ.
**
** Added by Stephen Frost <sfrost@snowman.net> to support
** having each runner process every N'th domain instead of
** every N'th message.
**
** Parameters:
** skip -- number of domains in WorkQ to skip.
**
** Returns:
** total number of messages skipped.
**
** Side Effects:
** may change WorkQ
*/
static int
skip_domains(skip)
int skip;
{
int n, seqjump;
for (n = 0, seqjump = 0; n < skip && WorkQ != NULL; seqjump++)
{
if (WorkQ->w_next != NULL)
{
if (WorkQ->w_host != NULL &&
WorkQ->w_next->w_host != NULL)
{
if (sm_strcasecmp(WorkQ->w_host,
WorkQ->w_next->w_host) != 0)
n++;
}
else
{
if ((WorkQ->w_host != NULL &&
WorkQ->w_next->w_host == NULL) ||
(WorkQ->w_host == NULL &&
WorkQ->w_next->w_host != NULL))
n++;
}
}
WorkQ = WorkQ->w_next;
}
return seqjump;
}
#endif /* _FFR_SKIP_DOMAINS */
/*
** RUNNER_WORK -- have a queue runner do its work
**
** Have a queue runner do its work a list of entries.
** When work isn't directly being done then this process can take a signal
** and terminate immediately (in a clean fashion of course).
** When work is directly being done, it's not to be interrupted
** immediately: the work should be allowed to finish at a clean point
** before termination (in a clean fashion of course).
**
** Parameters:
** e -- envelope.
** sequenceno -- 'th process to run WorkQ.
** didfork -- did the calling process fork()?
** skip -- process only each skip'th item.
** njobs -- number of jobs in WorkQ.
**
** Returns:
** none.
**
** Side Effects:
** runs things in the mail queue.
*/
static void
runner_work(e, sequenceno, didfork, skip, njobs)
register ENVELOPE *e;
int sequenceno;
bool didfork;
int skip;
int njobs;
{
int n, seqjump;
WORK *w;
time_t now;
SM_GET_LA(now);
/*
** Here we temporarily block the second calling of the handlers.
** This allows us to handle the signal without terminating in the
** middle of direct work. If a signal does come, the test for
** NoMoreRunners will find it.
*/
BlockOldsh = true;
seqjump = skip;
/* process them once at a time */
while (WorkQ != NULL)
{
#if SM_HEAP_CHECK
SM_NONVOLATILE int oldgroup = 0;
if (sm_debug_active(&DebugLeakQ, 1))
{
oldgroup = sm_heap_group();
sm_heap_newgroup();
sm_dprintf("run_queue_group() heap group #%d\n",
sm_heap_group());
}
#endif /* SM_HEAP_CHECK */
/* do no more work */
if (NoMoreRunners)
{
/* Check that a valid signal handler is callable */
if (Oldsh != SIG_DFL && Oldsh != SIG_IGN &&
Oldsh != runners_sighup &&
Oldsh != runners_sigterm)
(*Oldsh)(Oldsig);
break;
}
w = WorkQ; /* assign current work item */
/*
** Set the head of the WorkQ to the next work item.
** It is set 'skip' ahead (the number of parallel queue
** runners working on WorkQ together) since each runner
** works on every 'skip'th (N-th) item.
#if _FFR_SKIP_DOMAINS
** In the case of the BYHOST Queue Sort Order, the 'item'
** is a domain, so we work on every 'skip'th (N-th) domain.
#endif * _FFR_SKIP_DOMAINS *
*/
#if _FFR_SKIP_DOMAINS
if (QueueSortOrder == QSO_BYHOST)
{
seqjump = 1;
if (WorkQ->w_next != NULL)
{
if (WorkQ->w_host != NULL &&
WorkQ->w_next->w_host != NULL)
{
if (sm_strcasecmp(WorkQ->w_host,
WorkQ->w_next->w_host)
!= 0)
seqjump = skip_domains(skip);
else
WorkQ = WorkQ->w_next;
}
else
{
if ((WorkQ->w_host != NULL &&
WorkQ->w_next->w_host == NULL) ||
(WorkQ->w_host == NULL &&
WorkQ->w_next->w_host != NULL))
seqjump = skip_domains(skip);
else
WorkQ = WorkQ->w_next;
}
}
else
WorkQ = WorkQ->w_next;
}
else
#endif /* _FFR_SKIP_DOMAINS */
{
for (n = 0; n < skip && WorkQ != NULL; n++)
WorkQ = WorkQ->w_next;
}
e->e_to = NULL;
/*
** Ignore jobs that are too expensive for the moment.
**
** Get new load average every GET_NEW_LA_TIME seconds.
*/
SM_GET_LA(now);
if (shouldqueue(WkRecipFact, Current_LA_time))
{
char *msg = "Aborting queue run: load average too high";
if (Verbose)
message("%s", msg);
if (LogLevel > 8)
sm_syslog(LOG_INFO, NOQID, "runqueue: %s", msg);
break;
}
if (shouldqueue(w->w_pri, w->w_ctime))
{
if (Verbose)
message(EmptyString);
if (QueueSortOrder == QSO_BYPRIORITY)
{
if (Verbose)
message("Skipping %s/%s (sequence %d of %d) and flushing rest of queue",
qid_printqueue(w->w_qgrp,
w->w_qdir),
w->w_name + 2, sequenceno,
njobs);
if (LogLevel > 8)
sm_syslog(LOG_INFO, NOQID,
"runqueue: Flushing queue from %s/%s (pri %ld, LA %d, %d of %d)",
qid_printqueue(w->w_qgrp,
w->w_qdir),
w->w_name + 2, w->w_pri,
CurrentLA, sequenceno,
njobs);
break;
}
else if (Verbose)
message("Skipping %s/%s (sequence %d of %d)",
qid_printqueue(w->w_qgrp, w->w_qdir),
w->w_name + 2, sequenceno, njobs);
}
else
{
if (Verbose)
{
message(EmptyString);
message("Running %s/%s (sequence %d of %d)",
qid_printqueue(w->w_qgrp, w->w_qdir),
w->w_name + 2, sequenceno, njobs);
}
if (didfork && MaxQueueChildren > 0)
{
sm_blocksignal(SIGCHLD);
(void) sm_signal(SIGCHLD, reapchild);
}
if (tTd(63, 100))
sm_syslog(LOG_DEBUG, NOQID,
"runqueue %s dowork(%s)",
qid_printqueue(w->w_qgrp, w->w_qdir),
w->w_name + 2);
(void) dowork(w->w_qgrp, w->w_qdir, w->w_name + 2,
ForkQueueRuns, false, e);
errno = 0;
}
sm_free(w->w_name); /* XXX */
if (w->w_host != NULL)
sm_free(w->w_host); /* XXX */
sm_free((char *) w); /* XXX */
sequenceno += seqjump; /* next sequence number */
#if SM_HEAP_CHECK
if (sm_debug_active(&DebugLeakQ, 1))
sm_heap_setgroup(oldgroup);
#endif /* SM_HEAP_CHECK */
}
BlockOldsh = false;
/* check the signals didn't happen during the revert */
if (NoMoreRunners)
{
/* Check that a valid signal handler is callable */
if (Oldsh != SIG_DFL && Oldsh != SIG_IGN &&
Oldsh != runners_sighup && Oldsh != runners_sigterm)
(*Oldsh)(Oldsig);
}
Oldsh = SIG_DFL; /* after the NoMoreRunners check */
}
/*
** RUN_WORK_GROUP -- run the jobs in a queue group from a work group.
**
** Gets the stuff out of the queue in some presumably logical
** order and processes them.
**
** Parameters:
** wgrp -- work group to process.
** flags -- RWG_* flags
**
** Returns:
** true if the queue run successfully began.
**
** Side Effects:
** runs things in the mail queue.
*/
/* Minimum sleep time for persistent queue runners */
#define MIN_SLEEP_TIME 5
bool
run_work_group(wgrp, flags)
int wgrp;
int flags;
{
register ENVELOPE *e;
int njobs, qdir;
int sequenceno = 1;
int qgrp, endgrp, h, i;
time_t now;
bool full, more;
SM_RPOOL_T *rpool;
extern void rmexpstab __P((void));
extern ENVELOPE BlankEnvelope;
extern SIGFUNC_DECL reapchild __P((int));
if (wgrp < 0)
return false;
/*
** If no work will ever be selected, don't even bother reading
** the queue.
*/
SM_GET_LA(now);
if (!bitset(RWG_PERSISTENT, flags) &&
shouldqueue(WkRecipFact, Current_LA_time))
{
char *msg = "Skipping queue run -- load average too high";
if (bitset(RWG_VERBOSE, flags))
message("458 %s\n", msg);
if (LogLevel > 8)
sm_syslog(LOG_INFO, NOQID, "runqueue: %s", msg);
return false;
}
/*
** See if we already have too many children.
*/
if (bitset(RWG_FORK, flags) &&
WorkGrp[wgrp].wg_lowqintvl > 0 &&
!bitset(RWG_PERSISTENT, flags) &&
MaxChildren > 0 && CurChildren >= MaxChildren)
{
char *msg = "Skipping queue run -- too many children";
if (bitset(RWG_VERBOSE, flags))
message("458 %s (%d)\n", msg, CurChildren);
if (LogLevel > 8)
sm_syslog(LOG_INFO, NOQID, "runqueue: %s (%d)",
msg, CurChildren);
return false;
}
/*
** See if we want to go off and do other useful work.
*/
if (bitset(RWG_FORK, flags))
{
pid_t pid;
(void) sm_blocksignal(SIGCHLD);
(void) sm_signal(SIGCHLD, reapchild);
pid = dofork();
if (pid == -1)
{
const char *msg = "Skipping queue run -- fork() failed";
const char *err = sm_errstring(errno);
if (bitset(RWG_VERBOSE, flags))
message("458 %s: %s\n", msg, err);
if (LogLevel > 8)
sm_syslog(LOG_INFO, NOQID, "runqueue: %s: %s",
msg, err);
(void) sm_releasesignal(SIGCHLD);
return false;
}
if (pid != 0)
{
/* parent -- pick up intermediate zombie */
(void) sm_blocksignal(SIGALRM);
/* wgrp only used when queue runners are persistent */
proc_list_add(pid, "Queue runner", PROC_QUEUE,
WorkGrp[wgrp].wg_maxact,
bitset(RWG_PERSISTENT, flags) ? wgrp : -1,
NULL);
(void) sm_releasesignal(SIGALRM);
(void) sm_releasesignal(SIGCHLD);
return true;
}
/* child -- clean up signals */
/* Reset global flags */
RestartRequest = NULL;
RestartWorkGroup = false;
ShutdownRequest = NULL;
PendingSignal = 0;
CurrentPid = getpid();
close_sendmail_pid();
/*
** Initialize exception stack and default exception
** handler for child process.
*/
sm_exc_newthread(fatal_error);
clrcontrol();
proc_list_clear();
/* Add parent process as first child item */
proc_list_add(CurrentPid, "Queue runner child process",
PROC_QUEUE_CHILD, 0, -1, NULL);
(void) sm_releasesignal(SIGCHLD);
(void) sm_signal(SIGCHLD, SIG_DFL);
(void) sm_signal(SIGHUP, SIG_DFL);
(void) sm_signal(SIGTERM, intsig);
}
/*
** Release any resources used by the daemon code.
*/
clrdaemon();
/* force it to run expensive jobs */
NoConnect = false;
/* drop privileges */
if (geteuid() == (uid_t) 0)
(void) drop_privileges(false);
/*
** Create ourselves an envelope
*/
CurEnv = &QueueEnvelope;
rpool = sm_rpool_new_x(NULL);
e = newenvelope(&QueueEnvelope, CurEnv, rpool);
e->e_flags = BlankEnvelope.e_flags;
e->e_parent = NULL;
/* make sure we have disconnected from parent */
if (bitset(RWG_FORK, flags))
{
disconnect(1, e);
QuickAbort = false;
}
/*
** If we are running part of the queue, always ignore stored
** host status.
*/
if (QueueLimitId != NULL || QueueLimitSender != NULL ||
QueueLimitQuarantine != NULL ||
QueueLimitRecipient != NULL)
{
IgnoreHostStatus = true;
MinQueueAge = 0;
}
/*
** Here is where we choose the queue group from the work group.
** The caller of the "domorework" label must setup a new envelope.
*/
endgrp = WorkGrp[wgrp].wg_curqgrp; /* to not spin endlessly */
domorework:
/*
** Run a queue group if:
** RWG_RUNALL bit is set or the bit for this group is set.
*/
now = curtime();
for (;;)
{
/*
** Find the next queue group within the work group that
** has been marked as needing a run.
*/
qgrp = WorkGrp[wgrp].wg_qgs[WorkGrp[wgrp].wg_curqgrp]->qg_index;
WorkGrp[wgrp].wg_curqgrp++; /* advance */
WorkGrp[wgrp].wg_curqgrp %= WorkGrp[wgrp].wg_numqgrp; /* wrap */
if (bitset(RWG_RUNALL, flags) ||
(Queue[qgrp]->qg_nextrun <= now &&
Queue[qgrp]->qg_nextrun != (time_t) -1))
break;
if (endgrp == WorkGrp[wgrp].wg_curqgrp)
{
e->e_id = NULL;
if (bitset(RWG_FORK, flags))
finis(true, true, ExitStat);
return true; /* we're done */
}
}
qdir = Queue[qgrp]->qg_curnum; /* round-robin init of queue position */
#if _FFR_QUEUE_SCHED_DBG
if (tTd(69, 12))
sm_syslog(LOG_INFO, NOQID,
"rwg: wgrp=%d, qgrp=%d, qdir=%d, name=%s, curqgrp=%d, numgrps=%d",
wgrp, qgrp, qdir, qid_printqueue(qgrp, qdir),
WorkGrp[wgrp].wg_curqgrp, WorkGrp[wgrp].wg_numqgrp);
#endif /* _FFR_QUEUE_SCHED_DBG */
#if HASNICE
/* tweak niceness of queue runs */
if (Queue[qgrp]->qg_nice > 0)
(void) nice(Queue[qgrp]->qg_nice);
#endif /* HASNICE */
/* XXX running queue group... */
sm_setproctitle(true, CurEnv, "running queue: %s",
qid_printqueue(qgrp, qdir));
if (LogLevel > 69 || tTd(63, 99))
sm_syslog(LOG_DEBUG, NOQID,
"runqueue %s, pid=%d, forkflag=%d",
qid_printqueue(qgrp, qdir), (int) CurrentPid,
bitset(RWG_FORK, flags));
/*
** Start making passes through the queue.
** First, read and sort the entire queue.
** Then, process the work in that order.
** But if you take too long, start over.
*/
for (i = 0; i < Queue[qgrp]->qg_numqueues; i++)
{
h = gatherq(qgrp, qdir, false, &full, &more);
#if SM_CONF_SHM
if (ShmId != SM_SHM_NO_ID)
QSHM_ENTRIES(Queue[qgrp]->qg_qpaths[qdir].qp_idx) = h;
#endif /* SM_CONF_SHM */
/* If there are no more items in this queue advance */
if (!more)
{
/* A round-robin advance */
qdir++;
qdir %= Queue[qgrp]->qg_numqueues;
}
/* Has the WorkList reached the limit? */
if (full)
break; /* don't try to gather more */
}
/* order the existing work requests */
njobs = sortq(Queue[qgrp]->qg_maxlist);
Queue[qgrp]->qg_curnum = qdir; /* update */
if (!Verbose && bitnset(QD_FORK, Queue[qgrp]->qg_flags))
{
int loop, maxrunners;
pid_t pid;
/*
** For this WorkQ we want to fork off N children (maxrunners)
** at this point. Each child has a copy of WorkQ. Each child
** will process every N-th item. The parent will wait for all
** of the children to finish before moving on to the next
** queue group within the work group. This saves us forking
** a new runner-child for each work item.
** It's valid for qg_maxqrun == 0 since this may be an
** explicit "don't run this queue" setting.
*/
maxrunners = Queue[qgrp]->qg_maxqrun;
/* No need to have more runners then there are jobs */
if (maxrunners > njobs)
maxrunners = njobs;
for (loop = 0; loop < maxrunners; loop++)
{
/*
** Since the delivery may happen in a child and the
** parent does not wait, the parent may close the
** maps thereby removing any shared memory used by
** the map. Therefore, close the maps now so the
** child will dynamically open them if necessary.
*/
closemaps(false);
pid = fork();
if (pid < 0)
{
syserr("run_work_group: cannot fork");
return false;
}
else if (pid > 0)
{
/* parent -- clean out connection cache */
mci_flush(false, NULL);
#if _FFR_SKIP_DOMAINS
if (QueueSortOrder == QSO_BYHOST)
{
sequenceno += skip_domains(1);
}
else
#endif /* _FFR_SKIP_DOMAINS */
{
/* for the skip */
WorkQ = WorkQ->w_next;
sequenceno++;
}
proc_list_add(pid, "Queue child runner process",
PROC_QUEUE_CHILD, 0, -1, NULL);
/* No additional work, no additional runners */
if (WorkQ == NULL)
break;
}
else
{
/* child -- Reset global flags */
RestartRequest = NULL;
RestartWorkGroup = false;
ShutdownRequest = NULL;
PendingSignal = 0;
CurrentPid = getpid();
close_sendmail_pid();
/*
** Initialize exception stack and default
** exception handler for child process.
** When fork()'d the child now has a private
** copy of WorkQ at its current position.
*/
sm_exc_newthread(fatal_error);
/*
** SMTP processes (whether -bd or -bs) set
** SIGCHLD to reapchild to collect
** children status. However, at delivery
** time, that status must be collected
** by sm_wait() to be dealt with properly
** (check success of delivery based
** on status code, etc). Therefore, if we
** are an SMTP process, reset SIGCHLD
** back to the default so reapchild
** doesn't collect status before
** sm_wait().
*/
if (OpMode == MD_SMTP ||
OpMode == MD_DAEMON ||
MaxQueueChildren > 0)
{
proc_list_clear();
sm_releasesignal(SIGCHLD);
(void) sm_signal(SIGCHLD, SIG_DFL);
}
/* child -- error messages to the transcript */
QuickAbort = OnlyOneError = false;
runner_work(e, sequenceno, true,
maxrunners, njobs);
/* This child is done */
finis(true, true, ExitStat);
/* NOTREACHED */
}
}
sm_releasesignal(SIGCHLD);
/*
** Wait until all of the runners have completed before
** seeing if there is another queue group in the
** work group to process.
** XXX Future enhancement: don't wait() for all children
** here, just go ahead and make sure that overall the number
** of children is not exceeded.
*/
while (CurChildren > 0)
{
int status;
pid_t ret;
while ((ret = sm_wait(&status)) <= 0)
continue;
proc_list_drop(ret, status, NULL);
}
}
else if (Queue[qgrp]->qg_maxqrun > 0 || bitset(RWG_FORCE, flags))
{
/*
** When current process will not fork children to do the work,
** it will do the work itself. The 'skip' will be 1 since
** there are no child runners to divide the work across.
*/
runner_work(e, sequenceno, false, 1, njobs);
}
/* free memory allocated by newenvelope() above */
sm_rpool_free(rpool);
QueueEnvelope.e_rpool = NULL;
/* Are there still more queues in the work group to process? */
if (endgrp != WorkGrp[wgrp].wg_curqgrp)
{
rpool = sm_rpool_new_x(NULL);
e = newenvelope(&QueueEnvelope, CurEnv, rpool);
e->e_flags = BlankEnvelope.e_flags;
goto domorework;
}
/* No more queues in work group to process. Now check persistent. */
if (bitset(RWG_PERSISTENT, flags))
{
sequenceno = 1;
sm_setproctitle(true, CurEnv, "running queue: %s",
qid_printqueue(qgrp, qdir));
/*
** close bogus maps, i.e., maps which caused a tempfail,
** so we get fresh map connections on the next lookup.
** closemaps() is also called when children are started.
*/
closemaps(true);
/* Close any cached connections. */
mci_flush(true, NULL);
/* Clean out expired related entries. */
rmexpstab();
#if NAMED_BIND
/* Update MX records for FallbackMX. */
if (FallbackMX != NULL)
(void) getfallbackmxrr(FallbackMX);
#endif /* NAMED_BIND */
#if USERDB
/* close UserDatabase */
_udbx_close();
#endif /* USERDB */
#if SM_HEAP_CHECK
if (sm_debug_active(&SmHeapCheck, 2)
&& access("memdump", F_OK) == 0
)
{
SM_FILE_T *out;
remove("memdump");
out = sm_io_open(SmFtStdio, SM_TIME_DEFAULT,
"memdump.out", SM_IO_APPEND, NULL);
if (out != NULL)
{
(void) sm_io_fprintf(out, SM_TIME_DEFAULT, "----------------------\n");
sm_heap_report(out,
sm_debug_level(&SmHeapCheck) - 1);
(void) sm_io_close(out, SM_TIME_DEFAULT);
}
}
#endif /* SM_HEAP_CHECK */
/* let me rest for a second to catch my breath */
if (njobs == 0 && WorkGrp[wgrp].wg_lowqintvl < MIN_SLEEP_TIME)
sleep(MIN_SLEEP_TIME);
else if (WorkGrp[wgrp].wg_lowqintvl <= 0)
sleep(QueueIntvl > 0 ? QueueIntvl : MIN_SLEEP_TIME);
else
sleep(WorkGrp[wgrp].wg_lowqintvl);
/*
** Get the LA outside the WorkQ loop if necessary.
** In a persistent queue runner the code is repeated over
** and over but gatherq() may ignore entries due to
** shouldqueue() (do we really have to do this twice?).
** Hence the queue runners would just idle around when once
** CurrentLA caused all entries in a queue to be ignored.
*/
if (njobs == 0)
SM_GET_LA(now);
rpool = sm_rpool_new_x(NULL);
e = newenvelope(&QueueEnvelope, CurEnv, rpool);
e->e_flags = BlankEnvelope.e_flags;
goto domorework;
}
/* exit without the usual cleanup */
e->e_id = NULL;
if (bitset(RWG_FORK, flags))
finis(true, true, ExitStat);
/* NOTREACHED */
return true;
}
/*
** DOQUEUERUN -- do a queue run?
*/
bool
doqueuerun()
{
return DoQueueRun;
}
/*
** RUNQUEUEEVENT -- Sets a flag to indicate that a queue run should be done.
**
** Parameters:
** none.
**
** Returns:
** none.
**
** Side Effects:
** The invocation of this function via an alarm may interrupt
** a set of actions. Thus errno may be set in that context.
** We need to restore errno at the end of this function to ensure
** that any work done here that sets errno doesn't return a
** misleading/false errno value. Errno may be EINTR upon entry to
** this function because of non-restartable/continuable system
** API was active. Iff this is true we will override errno as
** a timeout (as a more accurate error message).
**
** NOTE: THIS CAN BE CALLED FROM A SIGNAL HANDLER. DO NOT ADD
** ANYTHING TO THIS ROUTINE UNLESS YOU KNOW WHAT YOU ARE
** DOING.
*/
void
runqueueevent(ignore)
int ignore;
{
int save_errno = errno;
/*
** Set the general bit that we want a queue run,
** tested in doqueuerun()
*/
DoQueueRun = true;
#if _FFR_QUEUE_SCHED_DBG
if (tTd(69, 10))
sm_syslog(LOG_INFO, NOQID, "rqe: done");
#endif /* _FFR_QUEUE_SCHED_DBG */
errno = save_errno;
if (errno == EINTR)
errno = ETIMEDOUT;
}
/*
** GATHERQ -- gather messages from the message queue(s) the work queue.
**
** Parameters:
** qgrp -- the index of the queue group.
** qdir -- the index of the queue directory.
** doall -- if set, include everything in the queue (even
** the jobs that cannot be run because the load
** average is too high, or MaxQueueRun is reached).
** Otherwise, exclude those jobs.
** full -- (optional) to be set 'true' if WorkList is full
** more -- (optional) to be set 'true' if there are still more
** messages in this queue not added to WorkList
**
** Returns:
** The number of request in the queue (not necessarily
** the number of requests in WorkList however).
**
** Side Effects:
** prepares available work into WorkList
*/
#define NEED_P 0001 /* 'P': priority */
#define NEED_T 0002 /* 'T': time */
#define NEED_R 0004 /* 'R': recipient */
#define NEED_S 0010 /* 'S': sender */
#define NEED_H 0020 /* host */
#define HAS_QUARANTINE 0040 /* has an unexpected 'q' line */
#define NEED_QUARANTINE 0100 /* 'q': reason */
static WORK *WorkList = NULL; /* list of unsort work */
static int WorkListSize = 0; /* current max size of WorkList */
static int WorkListCount = 0; /* # of work items in WorkList */
static int
gatherq(qgrp, qdir, doall, full, more)
int qgrp;
int qdir;
bool doall;
bool *full;
bool *more;
{
register struct dirent *d;
register WORK *w;
register char *p;
DIR *f;
int i, num_ent;
int wn;
QUEUE_CHAR *check;
char qd[MAXPATHLEN];
char qf[MAXPATHLEN];
wn = WorkListCount - 1;
num_ent = 0;
if (qdir == NOQDIR)
(void) sm_strlcpy(qd, ".", sizeof qd);
else
(void) sm_strlcpyn(qd, sizeof qd, 2,
Queue[qgrp]->qg_qpaths[qdir].qp_name,
(bitset(QP_SUBQF,
Queue[qgrp]->qg_qpaths[qdir].qp_subdirs)
? "/qf" : ""));
if (tTd(41, 1))
{
sm_dprintf("gatherq:\n");
check = QueueLimitId;
while (check != NULL)
{
sm_dprintf("\tQueueLimitId = %s%s\n",
check->queue_negate ? "!" : "",
check->queue_match);
check = check->queue_next;
}
check = QueueLimitSender;
while (check != NULL)
{
sm_dprintf("\tQueueLimitSender = %s%s\n",
check->queue_negate ? "!" : "",
check->queue_match);
check = check->queue_next;
}
check = QueueLimitRecipient;
while (check != NULL)
{
sm_dprintf("\tQueueLimitRecipient = %s%s\n",
check->queue_negate ? "!" : "",
check->queue_match);
check = check->queue_next;
}
if (QueueMode == QM_QUARANTINE)
{
check = QueueLimitQuarantine;
while (check != NULL)
{
sm_dprintf("\tQueueLimitQuarantine = %s%s\n",
check->queue_negate ? "!" : "",
check->queue_match);
check = check->queue_next;
}
}
}
/* open the queue directory */
f = opendir(qd);
if (f == NULL)
{
syserr("gatherq: cannot open \"%s\"",
qid_printqueue(qgrp, qdir));
if (full != NULL)
*full = WorkListCount >= MaxQueueRun && MaxQueueRun > 0;
if (more != NULL)
*more = false;
return 0;
}
/*
** Read the work directory.
*/
while ((d = readdir(f)) != NULL)
{
SM_FILE_T *cf;
int qfver = 0;
char lbuf[MAXNAME + 1];
struct stat sbuf;
if (tTd(41, 50))
sm_dprintf("gatherq: checking %s..", d->d_name);
/* is this an interesting entry? */
if (!(((QueueMode == QM_NORMAL &&
d->d_name[0] == NORMQF_LETTER) ||
(QueueMode == QM_QUARANTINE &&
d->d_name[0] == QUARQF_LETTER) ||
(QueueMode == QM_LOST &&
d->d_name[0] == LOSEQF_LETTER)) &&
d->d_name[1] == 'f'))
{
if (tTd(41, 50))
sm_dprintf(" skipping\n");
continue;
}
if (tTd(41, 50))
sm_dprintf("\n");
if (strlen(d->d_name) >= MAXQFNAME)
{
if (Verbose)
(void) sm_io_fprintf(smioout, SM_TIME_DEFAULT,
"gatherq: %s too long, %d max characters\n",
d->d_name, MAXQFNAME);
if (LogLevel > 0)
sm_syslog(LOG_ALERT, NOQID,
"gatherq: %s too long, %d max characters",
d->d_name, MAXQFNAME);
continue;
}
check = QueueLimitId;
while (check != NULL)
{
if (strcontainedin(false, check->queue_match,
d->d_name) != check->queue_negate)
break;
else
check = check->queue_next;
}
if (QueueLimitId != NULL && check == NULL)
continue;
/* grow work list if necessary */
if (++wn >= MaxQueueRun && MaxQueueRun > 0)
{
if (wn == MaxQueueRun && LogLevel > 0)
sm_syslog(LOG_WARNING, NOQID,
"WorkList for %s maxed out at %d",
qid_printqueue(qgrp, qdir),
MaxQueueRun);
if (doall)
continue; /* just count entries */
break;
}
if (wn >= WorkListSize)
{
grow_wlist(qgrp, qdir);
if (wn >= WorkListSize)
continue;
}
SM_ASSERT(wn >= 0);
w = &WorkList[wn];
(void) sm_strlcpyn(qf, sizeof qf, 3, qd, "/", d->d_name);
if (stat(qf, &sbuf) < 0)
{
if (errno != ENOENT)
sm_syslog(LOG_INFO, NOQID,
"gatherq: can't stat %s/%s",
qid_printqueue(qgrp, qdir),
d->d_name);
wn--;
continue;
}
if (!bitset(S_IFREG, sbuf.st_mode))
{
/* Yikes! Skip it or we will hang on open! */
if (!((d->d_name[0] == DATAFL_LETTER ||
d->d_name[0] == NORMQF_LETTER ||
d->d_name[0] == QUARQF_LETTER ||
d->d_name[0] == LOSEQF_LETTER ||
d->d_name[0] == XSCRPT_LETTER) &&
d->d_name[1] == 'f' && d->d_name[2] == '\0'))
syserr("gatherq: %s/%s is not a regular file",
qid_printqueue(qgrp, qdir), d->d_name);
wn--;
continue;
}
/* avoid work if possible */
if ((QueueSortOrder == QSO_BYFILENAME ||
QueueSortOrder == QSO_BYMODTIME ||
QueueSortOrder == QSO_RANDOM) &&
QueueLimitQuarantine == NULL &&
QueueLimitSender == NULL &&
QueueLimitRecipient == NULL)
{
w->w_qgrp = qgrp;
w->w_qdir = qdir;
w->w_name = newstr(d->d_name);
w->w_host = NULL;
w->w_lock = w->w_tooyoung = false;
w->w_pri = 0;
w->w_ctime = 0;
w->w_mtime = sbuf.st_mtime;
++num_ent;
continue;
}
/* open control file */
cf = sm_io_open(SmFtStdio, SM_TIME_DEFAULT, qf, SM_IO_RDONLY_B,
NULL);
if (cf == NULL && OpMode != MD_PRINT)
{
/* this may be some random person sending hir msgs */
if (tTd(41, 2))
sm_dprintf("gatherq: cannot open %s: %s\n",
d->d_name, sm_errstring(errno));
errno = 0;
wn--;
continue;
}
w->w_qgrp = qgrp;
w->w_qdir = qdir;
w->w_name = newstr(d->d_name);
w->w_host = NULL;
if (cf != NULL)
{
w->w_lock = !lockfile(sm_io_getinfo(cf, SM_IO_WHAT_FD,
NULL),
w->w_name, NULL,
LOCK_SH|LOCK_NB);
}
w->w_tooyoung = false;
/* make sure jobs in creation don't clog queue */
w->w_pri = 0x7fffffff;
w->w_ctime = 0;
w->w_mtime = sbuf.st_mtime;
/* extract useful information */
i = NEED_P|NEED_T;
if (QueueSortOrder == QSO_BYHOST
#if _FFR_RHS
|| QueueSortOrder == QSO_BYSHUFFLE
#endif /* _FFR_RHS */
)
{
/* need w_host set for host sort order */
i |= NEED_H;
}
if (QueueLimitSender != NULL)
i |= NEED_S;
if (QueueLimitRecipient != NULL)
i |= NEED_R;
if (QueueLimitQuarantine != NULL)
i |= NEED_QUARANTINE;
while (cf != NULL && i != 0 &&
sm_io_fgets(cf, SM_TIME_DEFAULT, lbuf,
sizeof lbuf) != NULL)
{
int c;
time_t age;
p = strchr(lbuf, '\n');
if (p != NULL)
*p = '\0';
else
{
/* flush rest of overly long line */
while ((c = sm_io_getc(cf, SM_TIME_DEFAULT))
!= SM_IO_EOF && c != '\n')
continue;
}
switch (lbuf[0])
{
case 'V':
qfver = atoi(&lbuf[1]);
break;
case 'P':
w->w_pri = atol(&lbuf[1]);
i &= ~NEED_P;
break;
case 'T':
w->w_ctime = atol(&lbuf[1]);
i &= ~NEED_T;
break;
case 'q':
if (QueueMode != QM_QUARANTINE &&
QueueMode != QM_LOST)
{
if (tTd(41, 49))
sm_dprintf("%s not marked as quarantined but has a 'q' line\n",
w->w_name);
i |= HAS_QUARANTINE;
}
else if (QueueMode == QM_QUARANTINE)
{
if (QueueLimitQuarantine == NULL)
{
i &= ~NEED_QUARANTINE;
break;
}
p = &lbuf[1];
check = QueueLimitQuarantine;
while (check != NULL)
{
if (strcontainedin(false,
check->queue_match,
p) !=
check->queue_negate)
break;
else
check = check->queue_next;
}
if (check != NULL)
i &= ~NEED_QUARANTINE;
}
break;
case 'R':
if (w->w_host == NULL &&
(p = strrchr(&lbuf[1], '@')) != NULL)
{
#if _FFR_RHS
if (QueueSortOrder == QSO_BYSHUFFLE)
w->w_host = newstr(&p[1]);
else
#endif /* _FFR_RHS */
w->w_host = strrev(&p[1]);
makelower(w->w_host);
i &= ~NEED_H;
}
if (QueueLimitRecipient == NULL)
{
i &= ~NEED_R;
break;
}
if (qfver > 0)
{
p = strchr(&lbuf[1], ':');
if (p == NULL)
p = &lbuf[1];
else
++p; /* skip over ':' */
}
else
p = &lbuf[1];
check = QueueLimitRecipient;
while (check != NULL)
{
if (strcontainedin(true,
check->queue_match,
p) !=
check->queue_negate)
break;
else
check = check->queue_next;
}
if (check != NULL)
i &= ~NEED_R;
break;
case 'S':
check = QueueLimitSender;
while (check != NULL)
{
if (strcontainedin(true,
check->queue_match,
&lbuf[1]) !=
check->queue_negate)
break;
else
check = check->queue_next;
}
if (check != NULL)
i &= ~NEED_S;
break;
case 'K':
age = curtime() - (time_t) atol(&lbuf[1]);
if (age >= 0 && MinQueueAge > 0 &&
age < MinQueueAge)
w->w_tooyoung = true;
break;
case 'N':
if (atol(&lbuf[1]) == 0)
w->w_tooyoung = false;
break;
}
}
if (cf != NULL)
(void) sm_io_close(cf, SM_TIME_DEFAULT);
if ((!doall && (shouldqueue(w->w_pri, w->w_ctime) ||
w->w_tooyoung)) ||
bitset(HAS_QUARANTINE, i) ||
bitset(NEED_QUARANTINE, i) ||
bitset(NEED_R|NEED_S, i))
{
/* don't even bother sorting this job in */
if (tTd(41, 49))
sm_dprintf("skipping %s (%x)\n", w->w_name, i);
sm_free(w->w_name); /* XXX */
if (w->w_host != NULL)
sm_free(w->w_host); /* XXX */
wn--;
}
else
++num_ent;
}
(void) closedir(f);
wn++;
i = wn - WorkListCount;
WorkListCount += SM_MIN(num_ent, WorkListSize);
if (more != NULL)
*more = WorkListCount < wn;
if (full != NULL)
*full = (wn >= MaxQueueRun && MaxQueueRun > 0) ||
(WorkList == NULL && wn > 0);
return i;
}
/*
** SORTQ -- sort the work list
**
** First the old WorkQ is cleared away. Then the WorkList is sorted
** for all items so that important (higher sorting value) items are not
** trunctated off. Then the most important items are moved from
** WorkList to WorkQ. The lower count of 'max' or MaxListCount items
** are moved.
**
** Parameters:
** max -- maximum number of items to be placed in WorkQ
**
** Returns:
** the number of items in WorkQ
**
** Side Effects:
** WorkQ gets released and filled with new work. WorkList
** gets released. Work items get sorted in order.
*/
static int
sortq(max)
int max;
{
register int i; /* local counter */
register WORK *w; /* tmp item pointer */
int wc = WorkListCount; /* trim size for WorkQ */
if (WorkQ != NULL)
{
WORK *nw;
/* Clear out old WorkQ. */
for (w = WorkQ; w != NULL; w = nw)
{
nw = w->w_next;
sm_free(w->w_name); /* XXX */
if (w->w_host != NULL)
sm_free(w->w_host); /* XXX */
sm_free((char *) w); /* XXX */
}
WorkQ = NULL;
}
if (WorkList == NULL || wc <= 0)
return 0;
/*
** The sort now takes place using all of the items in WorkList.
** The list gets trimmed to the most important items after the sort.
** If the trim were to happen before the sort then one or more
** important items might get truncated off -- not what we want.
*/
if (QueueSortOrder == QSO_BYHOST)
{
/*
** Sort the work directory for the first time,
** based on host name, lock status, and priority.
*/
qsort((char *) WorkList, wc, sizeof *WorkList, workcmpf1);
/*
** If one message to host is locked, "lock" all messages
** to that host.
*/
i = 0;
while (i < wc)
{
if (!WorkList[i].w_lock)
{
i++;
continue;
}
w = &WorkList[i];
while (++i < wc)
{
if (WorkList[i].w_host == NULL &&
w->w_host == NULL)
WorkList[i].w_lock = true;
else if (WorkList[i].w_host != NULL &&
w->w_host != NULL &&
sm_strcasecmp(WorkList[i].w_host,
w->w_host) == 0)
WorkList[i].w_lock = true;
else
break;
}
}
/*
** Sort the work directory for the second time,
** based on lock status, host name, and priority.
*/
qsort((char *) WorkList, wc, sizeof *WorkList, workcmpf2);
}
else if (QueueSortOrder == QSO_BYTIME)
{
/*
** Simple sort based on submission time only.
*/
qsort((char *) WorkList, wc, sizeof *WorkList, workcmpf3);
}
else if (QueueSortOrder == QSO_BYFILENAME)
{
/*
** Sort based on queue filename.
*/
qsort((char *) WorkList, wc, sizeof *WorkList, workcmpf4);
}
else if (QueueSortOrder == QSO_RANDOM)
{
/*
** Sort randomly. To avoid problems with an instable sort,
** use a random index into the queue file name to start
** comparison.
*/
randi = get_rand_mod(MAXQFNAME);
if (randi < 2)
randi = 3;
qsort((char *) WorkList, wc, sizeof *WorkList, workcmpf5);
}
else if (QueueSortOrder == QSO_BYMODTIME)
{
/*
** Simple sort based on modification time of queue file.
** This puts the oldest items first.
*/
qsort((char *) WorkList, wc, sizeof *WorkList, workcmpf6);
}
#if _FFR_RHS
else if (QueueSortOrder == QSO_BYSHUFFLE)
{
/*
** Simple sort based on shuffled host name.
*/
init_shuffle_alphabet();
qsort((char *) WorkList, wc, sizeof *WorkList, workcmpf7);
}
#endif /* _FFR_RHS */
else if (QueueSortOrder == QSO_BYPRIORITY)
{
/*
** Simple sort based on queue priority only.
*/
qsort((char *) WorkList, wc, sizeof *WorkList, workcmpf0);
}
/* else don't sort at all */
/* Check if the per queue group item limit will be exceeded */
if (wc > max && max > 0)
wc = max;
/*
** Convert the work list into canonical form.
** Should be turning it into a list of envelopes here perhaps.
** Only take the most important items up to the per queue group
** maximum.
*/
for (i = wc; --i >= 0; )
{
w = (WORK *) xalloc(sizeof *w);
w->w_qgrp = WorkList[i].w_qgrp;
w->w_qdir = WorkList[i].w_qdir;
w->w_name = WorkList[i].w_name;
w->w_host = WorkList[i].w_host;
w->w_lock = WorkList[i].w_lock;
w->w_tooyoung = WorkList[i].w_tooyoung;
w->w_pri = WorkList[i].w_pri;
w->w_ctime = WorkList[i].w_ctime;
w->w_mtime = WorkList[i].w_mtime;
w->w_next = WorkQ;
WorkQ = w;
}
/* free the rest of the list */
for (i = WorkListCount; --i >= wc; )
{
sm_free(WorkList[i].w_name);
if (WorkList[i].w_host != NULL)
sm_free(WorkList[i].w_host);
}
if (WorkList != NULL)
sm_free(WorkList); /* XXX */
WorkList = NULL;
WorkListSize = 0;
WorkListCount = 0;
if (tTd(40, 1))
{
for (w = WorkQ; w != NULL; w = w->w_next)
{
if (w->w_host != NULL)
sm_dprintf("%22s: pri=%ld %s\n",
w->w_name, w->w_pri, w->w_host);
else
sm_dprintf("%32s: pri=%ld\n",
w->w_name, w->w_pri);
}
}
return wc; /* return number of WorkQ items */
}
/*
** GROW_WLIST -- make the work list larger
**
** Parameters:
** qgrp -- the index for the queue group.
** qdir -- the index for the queue directory.
**
** Returns:
** none.
**
** Side Effects:
** Adds another QUEUESEGSIZE entries to WorkList if possible.
** It can fail if there isn't enough memory, so WorkListSize
** should be checked again upon return.
*/
static void
grow_wlist(qgrp, qdir)
int qgrp;
int qdir;
{
if (tTd(41, 1))
sm_dprintf("grow_wlist: WorkListSize=%d\n", WorkListSize);
if (WorkList == NULL)
{
WorkList = (WORK *) xalloc((sizeof *WorkList) *
(QUEUESEGSIZE + 1));
WorkListSize = QUEUESEGSIZE;
}
else
{
int newsize = WorkListSize + QUEUESEGSIZE;
WORK *newlist = (WORK *) sm_realloc((char *) WorkList,
(unsigned) sizeof(WORK) * (newsize + 1));
if (newlist != NULL)
{
WorkListSize = newsize;
WorkList = newlist;
if (LogLevel > 1)
{
sm_syslog(LOG_INFO, NOQID,
"grew WorkList for %s to %d",
qid_printqueue(qgrp, qdir),
WorkListSize);
}
}
else if (LogLevel > 0)
{
sm_syslog(LOG_ALERT, NOQID,
"FAILED to grow WorkList for %s to %d",
qid_printqueue(qgrp, qdir), newsize);
}
}
if (tTd(41, 1))
sm_dprintf("grow_wlist: WorkListSize now %d\n", WorkListSize);
}
/*
** WORKCMPF0 -- simple priority-only compare function.
**
** Parameters:
** a -- the first argument.
** b -- the second argument.
**
** Returns:
** -1 if a < b
** 0 if a == b
** +1 if a > b
**
*/
static int
workcmpf0(a, b)
register WORK *a;
register WORK *b;
{
long pa = a->w_pri;
long pb = b->w_pri;
if (pa == pb)
return 0;
else if (pa > pb)
return 1;
else
return -1;
}
/*
** WORKCMPF1 -- first compare function for ordering work based on host name.
**
** Sorts on host name, lock status, and priority in that order.
**
** Parameters:
** a -- the first argument.
** b -- the second argument.
**
** Returns:
** <0 if a < b
** 0 if a == b
** >0 if a > b
**
*/
static int
workcmpf1(a, b)
register WORK *a;
register WORK *b;
{
int i;
/* host name */
if (a->w_host != NULL && b->w_host == NULL)
return 1;
else if (a->w_host == NULL && b->w_host != NULL)
return -1;
if (a->w_host != NULL && b->w_host != NULL &&
(i = sm_strcasecmp(a->w_host, b->w_host)) != 0)
return i;
/* lock status */
if (a->w_lock != b->w_lock)
return b->w_lock - a->w_lock;
/* job priority */
return workcmpf0(a, b);
}
/*
** WORKCMPF2 -- second compare function for ordering work based on host name.
**
** Sorts on lock status, host name, and priority in that order.
**
** Parameters:
** a -- the first argument.
** b -- the second argument.
**
** Returns:
** <0 if a < b
** 0 if a == b
** >0 if a > b
**
*/
static int
workcmpf2(a, b)
register WORK *a;
register WORK *b;
{
int i;
/* lock status */
if (a->w_lock != b->w_lock)
return a->w_lock - b->w_lock;
/* host name */
if (a->w_host != NULL && b->w_host == NULL)
return 1;
else if (a->w_host == NULL && b->w_host != NULL)
return -1;
if (a->w_host != NULL && b->w_host != NULL &&
(i = sm_strcasecmp(a->w_host, b->w_host)) != 0)
return i;
/* job priority */
return workcmpf0(a, b);
}
/*
** WORKCMPF3 -- simple submission-time-only compare function.
**
** Parameters:
** a -- the first argument.
** b -- the second argument.
**
** Returns:
** -1 if a < b
** 0 if a == b
** +1 if a > b
**
*/
static int
workcmpf3(a, b)
register WORK *a;
register WORK *b;
{
if (a->w_ctime > b->w_ctime)
return 1;
else if (a->w_ctime < b->w_ctime)
return -1;
else
return 0;
}
/*
** WORKCMPF4 -- compare based on file name
**
** Parameters:
** a -- the first argument.
** b -- the second argument.
**
** Returns:
** -1 if a < b
** 0 if a == b
** +1 if a > b
**
*/
static int
workcmpf4(a, b)
register WORK *a;
register WORK *b;
{
return strcmp(a->w_name, b->w_name);
}
/*
** WORKCMPF5 -- compare based on assigned random number
**
** Parameters:
** a -- the first argument (ignored).
** b -- the second argument (ignored).
**
** Returns:
** randomly 1/-1
*/
/* ARGSUSED0 */
static int
workcmpf5(a, b)
register WORK *a;
register WORK *b;
{
if (strlen(a->w_name) < randi || strlen(b->w_name) < randi)
return -1;
return a->w_name[randi] - b->w_name[randi];
}
/*
** WORKCMPF6 -- simple modification-time-only compare function.
**
** Parameters:
** a -- the first argument.
** b -- the second argument.
**
** Returns:
** -1 if a < b
** 0 if a == b
** +1 if a > b
**
*/
static int
workcmpf6(a, b)
register WORK *a;
register WORK *b;
{
if (a->w_mtime > b->w_mtime)
return 1;
else if (a->w_mtime < b->w_mtime)
return -1;
else
return 0;
}
#if _FFR_RHS
/*
** WORKCMPF7 -- compare function for ordering work based on shuffled host name.
**
** Sorts on lock status, host name, and priority in that order.
**
** Parameters:
** a -- the first argument.
** b -- the second argument.
**
** Returns:
** <0 if a < b
** 0 if a == b
** >0 if a > b
**
*/
static int
workcmpf7(a, b)
register WORK *a;
register WORK *b;
{
int i;
/* lock status */
if (a->w_lock != b->w_lock)
return a->w_lock - b->w_lock;
/* host name */
if (a->w_host != NULL && b->w_host == NULL)
return 1;
else if (a->w_host == NULL && b->w_host != NULL)
return -1;
if (a->w_host != NULL && b->w_host != NULL &&
(i = sm_strshufflecmp(a->w_host, b->w_host)) != 0)
return i;
/* job priority */
return workcmpf0(a, b);
}
#endif /* _FFR_RHS */
/*
** STRREV -- reverse string
**
** Returns a pointer to a new string that is the reverse of
** the string pointed to by fwd. The space for the new
** string is obtained using xalloc().
**
** Parameters:
** fwd -- the string to reverse.
**
** Returns:
** the reversed string.
*/
static char *
strrev(fwd)
char *fwd;
{
char *rev = NULL;
int len, cnt;
len = strlen(fwd);
rev = xalloc(len + 1);
for (cnt = 0; cnt < len; ++cnt)
rev[cnt] = fwd[len - cnt - 1];
rev[len] = '\0';
return rev;
}
#if _FFR_RHS
# define NASCII 128
# define NCHAR 256
static unsigned char ShuffledAlphabet[NCHAR];
void
init_shuffle_alphabet()
{
static bool init = false;
int i;
if (init)
return;
/* fill the ShuffledAlphabet */
for (i = 0; i < NASCII; i++)
ShuffledAlphabet[i] = i;
/* mix it */
for (i = 1; i < NASCII; i++)
{
register int j = get_random() % NASCII;
register int tmp;
tmp = ShuffledAlphabet[j];
ShuffledAlphabet[j] = ShuffledAlphabet[i];
ShuffledAlphabet[i] = tmp;
}
/* make it case insensitive */
for (i = 'A'; i <= 'Z'; i++)
ShuffledAlphabet[i] = ShuffledAlphabet[i + 'a' - 'A'];
/* fill the upper part */
for (i = 0; i < NASCII; i++)
ShuffledAlphabet[i + NASCII] = ShuffledAlphabet[i];
init = true;
}
static int
sm_strshufflecmp(a, b)
char *a;
char *b;
{
const unsigned char *us1 = (const unsigned char *) a;
const unsigned char *us2 = (const unsigned char *) b;
while (ShuffledAlphabet[*us1] == ShuffledAlphabet[*us2++])
{
if (*us1++ == '\0')
return 0;
}
return (ShuffledAlphabet[*us1] - ShuffledAlphabet[*--us2]);
}
#endif /* _FFR_RHS */
/*
** DOWORK -- do a work request.
**
** Parameters:
** qgrp -- the index of the queue group for the job.
** qdir -- the index of the queue directory for the job.
** id -- the ID of the job to run.
** forkflag -- if set, run this in background.
** requeueflag -- if set, reinstantiate the queue quickly.
** This is used when expanding aliases in the queue.
** If forkflag is also set, it doesn't wait for the
** child.
** e - the envelope in which to run it.
**
** Returns:
** process id of process that is running the queue job.
**
** Side Effects:
** The work request is satisfied if possible.
*/
pid_t
dowork(qgrp, qdir, id, forkflag, requeueflag, e)
int qgrp;
int qdir;
char *id;
bool forkflag;
bool requeueflag;
register ENVELOPE *e;
{
register pid_t pid;
SM_RPOOL_T *rpool;
if (tTd(40, 1))
sm_dprintf("dowork(%s/%s)\n", qid_printqueue(qgrp, qdir), id);
/*
** Fork for work.
*/
if (forkflag)
{
/*
** Since the delivery may happen in a child and the
** parent does not wait, the parent may close the
** maps thereby removing any shared memory used by
** the map. Therefore, close the maps now so the
** child will dynamically open them if necessary.
*/
closemaps(false);
pid = fork();
if (pid < 0)
{
syserr("dowork: cannot fork");
return 0;
}
else if (pid > 0)
{
/* parent -- clean out connection cache */
mci_flush(false, NULL);
}
else
{
/*
** Initialize exception stack and default exception
** handler for child process.
*/
/* Reset global flags */
RestartRequest = NULL;
RestartWorkGroup = false;
ShutdownRequest = NULL;
PendingSignal = 0;
CurrentPid = getpid();
sm_exc_newthread(fatal_error);
/*
** See note above about SMTP processes and SIGCHLD.
*/
if (OpMode == MD_SMTP ||
OpMode == MD_DAEMON ||
MaxQueueChildren > 0)
{
proc_list_clear();
sm_releasesignal(SIGCHLD);
(void) sm_signal(SIGCHLD, SIG_DFL);
}
/* child -- error messages to the transcript */
QuickAbort = OnlyOneError = false;
}
}
else
{
pid = 0;
}
if (pid == 0)
{
/*
** CHILD
** Lock the control file to avoid duplicate deliveries.
** Then run the file as though we had just read it.
** We save an idea of the temporary name so we
** can recover on interrupt.
*/
if (forkflag)
{
/* Reset global flags */
RestartRequest = NULL;
RestartWorkGroup = false;
ShutdownRequest = NULL;
PendingSignal = 0;
}
/* set basic modes, etc. */
sm_clear_events();
clearstats();
rpool = sm_rpool_new_x(NULL);
clearenvelope(e, false, rpool);
e->e_flags |= EF_QUEUERUN|EF_GLOBALERRS;
set_delivery_mode(SM_DELIVER, e);
e->e_errormode = EM_MAIL;
e->e_id = id;
e->e_qgrp = qgrp;
e->e_qdir = qdir;
GrabTo = UseErrorsTo = false;
ExitStat = EX_OK;
if (forkflag)
{
disconnect(1, e);
set_op_mode(MD_QUEUERUN);
}
sm_setproctitle(true, e, "%s from queue", qid_printname(e));
if (LogLevel > 76)
sm_syslog(LOG_DEBUG, e->e_id, "dowork, pid=%d",
(int) CurrentPid);
/* don't use the headers from sendmail.cf... */
e->e_header = NULL;
/* read the queue control file -- return if locked */
if (!readqf(e, false))
{
if (tTd(40, 4) && e->e_id != NULL)
sm_dprintf("readqf(%s) failed\n",
qid_printname(e));
e->e_id = NULL;
if (forkflag)
finis(false, true, EX_OK);
else
{
/* adding this frees 8 bytes */
clearenvelope(e, false, rpool);
/* adding this frees 12 bytes */
sm_rpool_free(rpool);
e->e_rpool = NULL;
return 0;
}
}
e->e_flags |= EF_INQUEUE;
eatheader(e, requeueflag, true);
if (requeueflag)
queueup(e, false, false);
/* do the delivery */
sendall(e, SM_DELIVER);
/* finish up and exit */
if (forkflag)
finis(true, true, ExitStat);
else
{
dropenvelope(e, true, false);
sm_rpool_free(rpool);
e->e_rpool = NULL;
}
}
e->e_id = NULL;
return pid;
}
/*
** DOWORKLIST -- process a list of envelopes as work requests
**
** Similar to dowork(), except that after forking, it processes an
** envelope and its siblings, treating each envelope as a work request.
**
** Parameters:
** el -- envelope to be processed including its siblings.
** forkflag -- if set, run this in background.
** requeueflag -- if set, reinstantiate the queue quickly.
** This is used when expanding aliases in the queue.
** If forkflag is also set, it doesn't wait for the
** child.
**
** Returns:
** process id of process that is running the queue job.
**
** Side Effects:
** The work request is satisfied if possible.
*/
pid_t
doworklist(el, forkflag, requeueflag)
ENVELOPE *el;
bool forkflag;
bool requeueflag;
{
register pid_t pid;
ENVELOPE *ei;
if (tTd(40, 1))
sm_dprintf("doworklist()\n");
/*
** Fork for work.
*/
if (forkflag)
{
/*
** Since the delivery may happen in a child and the
** parent does not wait, the parent may close the
** maps thereby removing any shared memory used by
** the map. Therefore, close the maps now so the
** child will dynamically open them if necessary.
*/
closemaps(false);
pid = fork();
if (pid < 0)
{
syserr("doworklist: cannot fork");
return 0;
}
else if (pid > 0)
{
/* parent -- clean out connection cache */
mci_flush(false, NULL);
}
else
{
/*
** Initialize exception stack and default exception
** handler for child process.
*/
/* Reset global flags */
RestartRequest = NULL;
RestartWorkGroup = false;
ShutdownRequest = NULL;
PendingSignal = 0;
CurrentPid = getpid();
sm_exc_newthread(fatal_error);
/*
** See note above about SMTP processes and SIGCHLD.
*/
if (OpMode == MD_SMTP ||
OpMode == MD_DAEMON ||
MaxQueueChildren > 0)
{
proc_list_clear();
sm_releasesignal(SIGCHLD);
(void) sm_signal(SIGCHLD, SIG_DFL);
}
/* child -- error messages to the transcript */
QuickAbort = OnlyOneError = false;
}
}
else
{
pid = 0;
}
if (pid != 0)
return pid;
/*
** IN CHILD
** Lock the control file to avoid duplicate deliveries.
** Then run the file as though we had just read it.
** We save an idea of the temporary name so we
** can recover on interrupt.
*/
if (forkflag)
{
/* Reset global flags */
RestartRequest = NULL;
RestartWorkGroup = false;
ShutdownRequest = NULL;
PendingSignal = 0;
}
/* set basic modes, etc. */
sm_clear_events();
clearstats();
GrabTo = UseErrorsTo = false;
ExitStat = EX_OK;
if (forkflag)
{
disconnect(1, el);
set_op_mode(MD_QUEUERUN);
}
if (LogLevel > 76)
sm_syslog(LOG_DEBUG, el->e_id, "doworklist, pid=%d",
(int) CurrentPid);
for (ei = el; ei != NULL; ei = ei->e_sibling)
{
ENVELOPE e;
SM_RPOOL_T *rpool;
if (WILL_BE_QUEUED(ei->e_sendmode))
continue;
else if (QueueMode != QM_QUARANTINE &&
ei->e_quarmsg != NULL)
continue;
rpool = sm_rpool_new_x(NULL);
clearenvelope(&e, true, rpool);
e.e_flags |= EF_QUEUERUN|EF_GLOBALERRS;
set_delivery_mode(SM_DELIVER, &e);
e.e_errormode = EM_MAIL;
e.e_id = ei->e_id;
e.e_qgrp = ei->e_qgrp;
e.e_qdir = ei->e_qdir;
openxscript(&e);
sm_setproctitle(true, &e, "%s from queue", qid_printname(&e));
/* don't use the headers from sendmail.cf... */
e.e_header = NULL;
CurEnv = &e;
/* read the queue control file -- return if locked */
if (readqf(&e, false))
{
e.e_flags |= EF_INQUEUE;
eatheader(&e, requeueflag, true);
if (requeueflag)
queueup(&e, false, false);
/* do the delivery */
sendall(&e, SM_DELIVER);
dropenvelope(&e, true, false);
}
else
{
if (tTd(40, 4) && e.e_id != NULL)
sm_dprintf("readqf(%s) failed\n",
qid_printname(&e));
}
sm_rpool_free(rpool);
ei->e_id = NULL;
}
/* restore CurEnv */
CurEnv = el;
/* finish up and exit */
if (forkflag)
finis(true, true, ExitStat);
return 0;
}
/*
** READQF -- read queue file and set up environment.
**
** Parameters:
** e -- the envelope of the job to run.
** openonly -- only open the qf (returned as e_lockfp)
**
** Returns:
** true if it successfully read the queue file.
** false otherwise.
**
** Side Effects:
** The queue file is returned locked.
*/
static bool
readqf(e, openonly)
register ENVELOPE *e;
bool openonly;
{
register SM_FILE_T *qfp;
ADDRESS *ctladdr;
struct stat st, stf;
char *bp;
int qfver = 0;
long hdrsize = 0;
register char *p;
char *frcpt = NULL;
char *orcpt = NULL;
bool nomore = false;
bool bogus = false;
MODE_T qsafe;
char *err;
char qf[MAXPATHLEN];
char buf[MAXLINE];
/*
** Read and process the file.
*/
bp = NULL;
(void) sm_strlcpy(qf, queuename(e, ANYQFL_LETTER), sizeof qf);
qfp = sm_io_open(SmFtStdio, SM_TIME_DEFAULT, qf, SM_IO_RDWR_B, NULL);
if (qfp == NULL)
{
int save_errno = errno;
if (tTd(40, 8))
sm_dprintf("readqf(%s): sm_io_open failure (%s)\n",
qf, sm_errstring(errno));
errno = save_errno;
if (errno != ENOENT
)
syserr("readqf: no control file %s", qf);
RELEASE_QUEUE;
return false;
}
if (!lockfile(sm_io_getinfo(qfp, SM_IO_WHAT_FD, NULL), qf, NULL,
LOCK_EX|LOCK_NB))
{
/* being processed by another queuer */
if (Verbose)
(void) sm_io_fprintf(smioout, SM_TIME_DEFAULT,
"%s: locked\n", e->e_id);
if (tTd(40, 8))
sm_dprintf("%s: locked\n", e->e_id);
if (LogLevel > 19)
sm_syslog(LOG_DEBUG, e->e_id, "locked");
(void) sm_io_close(qfp, SM_TIME_DEFAULT);
RELEASE_QUEUE;
return false;
}
RELEASE_QUEUE;
/*
** Prevent locking race condition.
**
** Process A: readqf(): qfp = fopen(qffile)
** Process B: queueup(): rename(tf, qf)
** Process B: unlocks(tf)
** Process A: lockfile(qf);
**
** Process A (us) has the old qf file (before the rename deleted
** the directory entry) and will be delivering based on old data.
** This can lead to multiple deliveries of the same recipients.
**
** Catch this by checking if the underlying qf file has changed
** *after* acquiring our lock and if so, act as though the file
** was still locked (i.e., just return like the lockfile() case
** above.
*/
if (stat(qf, &stf) < 0 ||
fstat(sm_io_getinfo(qfp, SM_IO_WHAT_FD, NULL), &st) < 0)
{
/* must have been being processed by someone else */
if (tTd(40, 8))
sm_dprintf("readqf(%s): [f]stat failure (%s)\n",
qf, sm_errstring(errno));
(void) sm_io_close(qfp, SM_TIME_DEFAULT);
return false;
}
if (st.st_nlink != stf.st_nlink ||
st.st_dev != stf.st_dev ||
ST_INODE(st) != ST_INODE(stf) ||
#if HAS_ST_GEN && 0 /* AFS returns garbage in st_gen */
st.st_gen != stf.st_gen ||
#endif /* HAS_ST_GEN && 0 */
st.st_uid != stf.st_uid ||
st.st_gid != stf.st_gid ||
st.st_size != stf.st_size)
{
/* changed after opened */
if (Verbose)
(void) sm_io_fprintf(smioout, SM_TIME_DEFAULT,
"%s: changed\n", e->e_id);
if (tTd(40, 8))
sm_dprintf("%s: changed\n", e->e_id);
if (LogLevel > 19)
sm_syslog(LOG_DEBUG, e->e_id, "changed");
(void) sm_io_close(qfp, SM_TIME_DEFAULT);
return false;
}
/*
** Check the queue file for plausibility to avoid attacks.
*/
qsafe = S_IWOTH|S_IWGRP;
if (bitset(S_IWGRP, QueueFileMode))
qsafe &= ~S_IWGRP;
bogus = st.st_uid != geteuid() &&
st.st_uid != TrustedUid &&
geteuid() != RealUid;
/*
** If this qf file results from a set-group-ID binary, then
** we check whether the directory is group-writable,
** the queue file mode contains the group-writable bit, and
** the groups are the same.
** Notice: this requires that the set-group-ID binary is used to
** run the queue!
*/
if (bogus && st.st_gid == getegid() && UseMSP)
{
char delim;
struct stat dst;
bp = SM_LAST_DIR_DELIM(qf);
if (bp == NULL)
delim = '\0';
else
{
delim = *bp;
*bp = '\0';
}
if (stat(delim == '\0' ? "." : qf, &dst) < 0)
syserr("readqf: cannot stat directory %s",
delim == '\0' ? "." : qf);
else
{
bogus = !(bitset(S_IWGRP, QueueFileMode) &&
bitset(S_IWGRP, dst.st_mode) &&
dst.st_gid == st.st_gid);
}
if (delim != '\0')
*bp = delim;
bp = NULL;
}
if (!bogus)
bogus = bitset(qsafe, st.st_mode);
if (bogus)
{
if (LogLevel > 0)
{
sm_syslog(LOG_ALERT, e->e_id,
"bogus queue file, uid=%d, gid=%d, mode=%o",
st.st_uid, st.st_gid, st.st_mode);
}
if (tTd(40, 8))
sm_dprintf("readqf(%s): bogus file\n", qf);
e->e_flags |= EF_INQUEUE;
if (!openonly)
loseqfile(e, "bogus file uid/gid in mqueue");
(void) sm_io_close(qfp, SM_TIME_DEFAULT);
return false;
}
if (st.st_size == 0)
{
/* must be a bogus file -- if also old, just remove it */
if (!openonly && st.st_ctime + 10 * 60 < curtime())
{
(void) xunlink(queuename(e, DATAFL_LETTER));
(void) xunlink(queuename(e, ANYQFL_LETTER));
}
(void) sm_io_close(qfp, SM_TIME_DEFAULT);
return false;
}
if (st.st_nlink == 0)
{
/*
** Race condition -- we got a file just as it was being
** unlinked. Just assume it is zero length.
*/
(void) sm_io_close(qfp, SM_TIME_DEFAULT);
return false;
}
#if _FFR_TRUSTED_QF
/*
** If we don't own the file mark it as unsafe.
** However, allow TrustedUser to own it as well
** in case TrustedUser manipulates the queue.
*/
if (st.st_uid != geteuid() && st.st_uid != TrustedUid)
e->e_flags |= EF_UNSAFE;
#else /* _FFR_TRUSTED_QF */
/* If we don't own the file mark it as unsafe */
if (st.st_uid != geteuid())
e->e_flags |= EF_UNSAFE;
#endif /* _FFR_TRUSTED_QF */
/* good file -- save this lock */
e->e_lockfp = qfp;
/* Just wanted the open file */
if (openonly)
return true;
/* do basic system initialization */
initsys(e);
macdefine(&e->e_macro, A_PERM, 'i', e->e_id);
LineNumber = 0;
e->e_flags |= EF_GLOBALERRS;
set_op_mode(MD_QUEUERUN);
ctladdr = NULL;
e->e_qfletter = queue_letter(e, ANYQFL_LETTER);
e->e_dfqgrp = e->e_qgrp;
e->e_dfqdir = e->e_qdir;
#if _FFR_QUEUE_MACRO
macdefine(&e->e_macro, A_TEMP, macid("{queue}"),
qid_printqueue(e->e_qgrp, e->e_qdir));
#endif /* _FFR_QUEUE_MACRO */
e->e_dfino = -1;
e->e_msgsize = -1;
while ((bp = fgetfolded(buf, sizeof buf, qfp)) != NULL)
{
unsigned long qflags;
ADDRESS *q;
int r;
time_t now;
auto char *ep;
if (tTd(40, 4))
sm_dprintf("+++++ %s\n", bp);
if (nomore)
{
/* hack attack */
hackattack:
syserr("SECURITY ALERT: extra or bogus data in queue file: %s",
bp);
err = "bogus queue line";
goto fail;
}
switch (bp[0])
{
case 'A': /* AUTH= parameter */
if (!xtextok(&bp[1]))
goto hackattack;
e->e_auth_param = sm_rpool_strdup_x(e->e_rpool, &bp[1]);
break;
case 'B': /* body type */
r = check_bodytype(&bp[1]);
if (!BODYTYPE_VALID(r))
goto hackattack;
e->e_bodytype = sm_rpool_strdup_x(e->e_rpool, &bp[1]);
break;
case 'C': /* specify controlling user */
ctladdr = setctluser(&bp[1], qfver, e);
break;
case 'D': /* data file name */
/* obsolete -- ignore */
break;
case 'd': /* data file directory name */
{
int qgrp, qdir;
#if _FFR_MSP_PARANOIA
/* forbid queue groups in MSP? */
if (UseMSP)
goto hackattack;
#endif /* _FFR_MSP_PARANOIA */
for (qgrp = 0;
qgrp < NumQueue && Queue[qgrp] != NULL;
++qgrp)
{
for (qdir = 0;
qdir < Queue[qgrp]->qg_numqueues;
++qdir)
{
if (strcmp(&bp[1],
Queue[qgrp]->qg_qpaths[qdir].qp_name)
== 0)
{
e->e_dfqgrp = qgrp;
e->e_dfqdir = qdir;
goto done;
}
}
}
err = "bogus queue file directory";
goto fail;
done:
break;
}
case 'E': /* specify error recipient */
/* no longer used */
break;
case 'F': /* flag bits */
if (strncmp(bp, "From ", 5) == 0)
{
/* we are being spoofed! */
syserr("SECURITY ALERT: bogus qf line %s", bp);
err = "bogus queue line";
goto fail;
}
for (p = &bp[1]; *p != '\0'; p++)
{
switch (*p)
{
case '8': /* has 8 bit data */
e->e_flags |= EF_HAS8BIT;
break;
case 'b': /* delete Bcc: header */
e->e_flags |= EF_DELETE_BCC;
break;
case 'd': /* envelope has DSN RET= */
e->e_flags |= EF_RET_PARAM;
break;
case 'n': /* don't return body */
e->e_flags |= EF_NO_BODY_RETN;
break;
case 'r': /* response */
e->e_flags |= EF_RESPONSE;
break;
case 's': /* split */
e->e_flags |= EF_SPLIT;
break;
case 'w': /* warning sent */
e->e_flags |= EF_WARNING;
break;
}
}
break;
case 'q': /* quarantine reason */
e->e_quarmsg = sm_rpool_strdup_x(e->e_rpool, &bp[1]);
macdefine(&e->e_macro, A_PERM,
macid("{quarantine}"), e->e_quarmsg);
break;
case 'H': /* header */
/*
** count size before chompheader() destroys the line.
** this isn't accurate due to macro expansion, but
** better than before. "-3" to skip H?? at least.
*/
hdrsize += strlen(bp) - 3;
(void) chompheader(&bp[1], CHHDR_QUEUE, NULL, e);
break;
case 'I': /* data file's inode number */
/* regenerated below */
break;
case 'K': /* time of last delivery attempt */
e->e_dtime = atol(&buf[1]);
break;
case 'L': /* Solaris Content-Length: */
case 'M': /* message */
/* ignore this; we want a new message next time */
break;
case 'N': /* number of delivery attempts */
e->e_ntries = atoi(&buf[1]);
/* if this has been tried recently, let it be */
now = curtime();
if (e->e_ntries > 0 && e->e_dtime <= now &&
now < e->e_dtime + MinQueueAge)
{
char *howlong;
howlong = pintvl(now - e->e_dtime, true);
if (Verbose)
(void) sm_io_fprintf(smioout,
SM_TIME_DEFAULT,
"%s: too young (%s)\n",
e->e_id, howlong);
if (tTd(40, 8))
sm_dprintf("%s: too young (%s)\n",
e->e_id, howlong);
if (LogLevel > 19)
sm_syslog(LOG_DEBUG, e->e_id,
"too young (%s)",
howlong);
e->e_id = NULL;
unlockqueue(e);
return false;
}
macdefine(&e->e_macro, A_TEMP,
macid("{ntries}"), &buf[1]);
#if NAMED_BIND
/* adjust BIND parameters immediately */
if (e->e_ntries == 0)
{
_res.retry = TimeOuts.res_retry[RES_TO_FIRST];
_res.retrans = TimeOuts.res_retrans[RES_TO_FIRST];
}
else
{
_res.retry = TimeOuts.res_retry[RES_TO_NORMAL];
_res.retrans = TimeOuts.res_retrans[RES_TO_NORMAL];
}
#endif /* NAMED_BIND */
break;
case 'P': /* message priority */
e->e_msgpriority = atol(&bp[1]) + WkTimeFact;
break;
case 'Q': /* original recipient */
orcpt = sm_rpool_strdup_x(e->e_rpool, &bp[1]);
break;
case 'r': /* final recipient */
frcpt = sm_rpool_strdup_x(e->e_rpool, &bp[1]);
break;
case 'R': /* specify recipient */
p = bp;
qflags = 0;
if (qfver >= 1)
{
/* get flag bits */
while (*++p != '\0' && *p != ':')
{
switch (*p)
{
case 'N':
qflags |= QHASNOTIFY;
break;
case 'S':
qflags |= QPINGONSUCCESS;
break;
case 'F':
qflags |= QPINGONFAILURE;
break;
case 'D':
qflags |= QPINGONDELAY;
break;
case 'P':
qflags |= QPRIMARY;
break;
case 'A':
if (ctladdr != NULL)
ctladdr->q_flags |= QALIAS;
break;
default: /* ignore or complain? */
break;
}
}
}
else
qflags |= QPRIMARY;
macdefine(&e->e_macro, A_PERM, macid("{addr_type}"),
"e r");
if (*p != '\0')
q = parseaddr(++p, NULLADDR, RF_COPYALL, '\0',
NULL, e, true);
else
q = NULL;
if (q != NULL)
{
/* make sure we keep the current qgrp */
if (ISVALIDQGRP(e->e_qgrp))
q->q_qgrp = e->e_qgrp;
q->q_alias = ctladdr;
if (qfver >= 1)
q->q_flags &= ~Q_PINGFLAGS;
q->q_flags |= qflags;
q->q_finalrcpt = frcpt;
q->q_orcpt = orcpt;
(void) recipient(q, &e->e_sendqueue, 0, e);
}
frcpt = NULL;
orcpt = NULL;
macdefine(&e->e_macro, A_PERM, macid("{addr_type}"),
NULL);
break;
case 'S': /* sender */
setsender(sm_rpool_strdup_x(e->e_rpool, &bp[1]),
e, NULL, '\0', true);
break;
case 'T': /* init time */
e->e_ctime = atol(&bp[1]);
break;
case 'V': /* queue file version number */
qfver = atoi(&bp[1]);
if (qfver <= QF_VERSION)
break;
syserr("Version number in queue file (%d) greater than max (%d)",
qfver, QF_VERSION);
err = "unsupported queue file version";
goto fail;
/* NOTREACHED */
break;
case 'Z': /* original envelope id from ESMTP */
e->e_envid = sm_rpool_strdup_x(e->e_rpool, &bp[1]);
macdefine(&e->e_macro, A_PERM,
macid("{dsn_envid}"), e->e_envid);
break;
case '!': /* deliver by */
/* format: flag (1 char) space long-integer */
e->e_dlvr_flag = buf[1];
e->e_deliver_by = strtol(&buf[3], NULL, 10);
case '$': /* define macro */
{
char *p;
/* XXX elimate p? */
r = macid_parse(&bp[1], &ep);
if (r == 0)
break;
p = sm_rpool_strdup_x(e->e_rpool, ep);
macdefine(&e->e_macro, A_PERM, r, p);
}
break;
case '.': /* terminate file */
nomore = true;
break;
#if _FFR_QUEUEDELAY
case 'G':
case 'Y':
/*
** Maintain backward compatibility for
** users who defined _FFR_QUEUEDELAY in
** previous releases. Remove this
** code in 8.14 or 8.15.
*/
if (qfver == 5 || qfver == 7)
break;
/* If not qfver 5 or 7, then 'G' or 'Y' is invalid */
/* FALLTHROUGH */
#endif /* _FFR_QUEUEDELAY */
default:
syserr("readqf: %s: line %d: bad line \"%s\"",
qf, LineNumber, shortenstring(bp, MAXSHORTSTR));
err = "unrecognized line";
goto fail;
}
if (bp != buf)
{
sm_free(bp); /* XXX */
bp = NULL;
}
}
/*
** If we haven't read any lines, this queue file is empty.
** Arrange to remove it without referencing any null pointers.
*/
if (LineNumber == 0)
{
errno = 0;
e->e_flags |= EF_CLRQUEUE|EF_FATALERRS|EF_RESPONSE;
return true;
}
/* Check to make sure we have a complete queue file read */
if (!nomore)
{
syserr("readqf: %s: incomplete queue file read", qf);
(void) sm_io_close(qfp, SM_TIME_DEFAULT);
return false;
}
/* possibly set ${dsn_ret} macro */
if (bitset(EF_RET_PARAM, e->e_flags))
{
if (bitset(EF_NO_BODY_RETN, e->e_flags))
macdefine(&e->e_macro, A_PERM,
macid("{dsn_ret}"), "hdrs");
else
macdefine(&e->e_macro, A_PERM,
macid("{dsn_ret}"), "full");
}
/*
** Arrange to read the data file.
*/
p = queuename(e, DATAFL_LETTER);
e->e_dfp = sm_io_open(SmFtStdio, SM_TIME_DEFAULT, p, SM_IO_RDONLY_B,
NULL);
if (e->e_dfp == NULL)
{
syserr("readqf: cannot open %s", p);
}
else
{
e->e_flags |= EF_HAS_DF;
if (fstat(sm_io_getinfo(e->e_dfp, SM_IO_WHAT_FD, NULL), &st)
>= 0)
{
e->e_msgsize = st.st_size + hdrsize;
e->e_dfdev = st.st_dev;
e->e_dfino = ST_INODE(st);
(void) sm_snprintf(buf, sizeof buf, "%ld",
e->e_msgsize);
macdefine(&e->e_macro, A_TEMP, macid("{msg_size}"),
buf);
}
}
return true;
fail:
/*
** There was some error reading the qf file (reason is in err var.)
** Cleanup:
** close file; clear e_lockfp since it is the same as qfp,
** hence it is invalid (as file) after qfp is closed;
** the qf file is on disk, so set the flag to avoid calling
** queueup() with bogus data.
*/
if (bp != NULL && bp != buf)
{
sm_free(bp); /* XXX */
bp = NULL;
}
if (qfp != NULL)
(void) sm_io_close(qfp, SM_TIME_DEFAULT);
e->e_lockfp = NULL;
e->e_flags |= EF_INQUEUE;
loseqfile(e, err);
return false;
}
/*
** PRTSTR -- print a string, "unprintable" characters are shown as \oct
**
** Parameters:
** s -- string to print
** ml -- maximum length of output
**
** Returns:
** number of entries
**
** Side Effects:
** Prints a string on stdout.
*/
static void
prtstr(s, ml)
char *s;
int ml;
{
int c;
if (s == NULL)
return;
while (ml-- > 0 && ((c = *s++) != '\0'))
{
if (c == '\\')
{
if (ml-- > 0)
{
(void) sm_io_putc(smioout, SM_TIME_DEFAULT, c);
(void) sm_io_putc(smioout, SM_TIME_DEFAULT, c);
}
}
else if (isascii(c) && isprint(c))
(void) sm_io_putc(smioout, SM_TIME_DEFAULT, c);
else
{
if ((ml -= 3) > 0)
(void) sm_io_fprintf(smioout, SM_TIME_DEFAULT,
"\\%03o", c & 0xFF);
}
}
}
/*
** PRINTNQE -- print out number of entries in the mail queue
**
** Parameters:
** out -- output file pointer.
** prefix -- string to output in front of each line.
**
** Returns:
** none.
*/
void
printnqe(out, prefix)
SM_FILE_T *out;
char *prefix;
{
#if SM_CONF_SHM
int i, k = 0, nrequests = 0;
bool unknown = false;
if (ShmId == SM_SHM_NO_ID)
{
if (prefix == NULL)
(void) sm_io_fprintf(out, SM_TIME_DEFAULT,
"Data unavailable: shared memory not updated\n");
else
(void) sm_io_fprintf(out, SM_TIME_DEFAULT,
"%sNOTCONFIGURED:-1\r\n", prefix);
return;
}
for (i = 0; i < NumQueue && Queue[i] != NULL; i++)
{
int j;
k++;
for (j = 0; j < Queue[i]->qg_numqueues; j++)
{
int n;
if (StopRequest)
stop_sendmail();
n = QSHM_ENTRIES(Queue[i]->qg_qpaths[j].qp_idx);
if (prefix != NULL)
(void) sm_io_fprintf(out, SM_TIME_DEFAULT,
"%s%s:%d\r\n",
prefix, qid_printqueue(i, j), n);
else if (n < 0)
{
(void) sm_io_fprintf(out, SM_TIME_DEFAULT,
"%s: unknown number of entries\n",
qid_printqueue(i, j));
unknown = true;
}
else if (n == 0)
{
(void) sm_io_fprintf(out, SM_TIME_DEFAULT,
"%s is empty\n",
qid_printqueue(i, j));
}
else if (n > 0)
{
(void) sm_io_fprintf(out, SM_TIME_DEFAULT,
"%s: entries=%d\n",
qid_printqueue(i, j), n);
nrequests += n;
k++;
}
}
}
if (prefix == NULL && k > 1)
(void) sm_io_fprintf(out, SM_TIME_DEFAULT,
"\t\tTotal requests: %d%s\n",
nrequests, unknown ? " (about)" : "");
#else /* SM_CONF_SHM */
if (prefix == NULL)
(void) sm_io_fprintf(out, SM_TIME_DEFAULT,
"Data unavailable without shared memory support\n");
else
(void) sm_io_fprintf(out, SM_TIME_DEFAULT,
"%sNOTAVAILABLE:-1\r\n", prefix);
#endif /* SM_CONF_SHM */
}
/*
** PRINTQUEUE -- print out a representation of the mail queue
**
** Parameters:
** none.
**
** Returns:
** none.
**
** Side Effects:
** Prints a listing of the mail queue on the standard output.
*/
void
printqueue()
{
int i, k = 0, nrequests = 0;
for (i = 0; i < NumQueue && Queue[i] != NULL; i++)
{
int j;
k++;
for (j = 0; j < Queue[i]->qg_numqueues; j++)
{
if (StopRequest)
stop_sendmail();
nrequests += print_single_queue(i, j);
k++;
}
}
if (k > 1)
(void) sm_io_fprintf(smioout, SM_TIME_DEFAULT,
"\t\tTotal requests: %d\n",
nrequests);
}
/*
** PRINT_SINGLE_QUEUE -- print out a representation of a single mail queue
**
** Parameters:
** qgrp -- the index of the queue group.
** qdir -- the queue directory.
**
** Returns:
** number of requests in mail queue.
**
** Side Effects:
** Prints a listing of the mail queue on the standard output.
*/
int
print_single_queue(qgrp, qdir)
int qgrp;
int qdir;
{
register WORK *w;
SM_FILE_T *f;
int nrequests;
char qd[MAXPATHLEN];
char qddf[MAXPATHLEN];
char buf[MAXLINE];
if (qdir == NOQDIR)
{
(void) sm_strlcpy(qd, ".", sizeof qd);
(void) sm_strlcpy(qddf, ".", sizeof qddf);
}
else
{
(void) sm_strlcpyn(qd, sizeof qd, 2,
Queue[qgrp]->qg_qpaths[qdir].qp_name,
(bitset(QP_SUBQF,
Queue[qgrp]->qg_qpaths[qdir].qp_subdirs)
? "/qf" : ""));
(void) sm_strlcpyn(qddf, sizeof qddf, 2,
Queue[qgrp]->qg_qpaths[qdir].qp_name,
(bitset(QP_SUBDF,
Queue[qgrp]->qg_qpaths[qdir].qp_subdirs)
? "/df" : ""));
}
/*
** Check for permission to print the queue
*/
if (bitset(PRIV_RESTRICTMAILQ, PrivacyFlags) && RealUid != 0)
{
struct stat st;
#ifdef NGROUPS_MAX
int n;
extern GIDSET_T InitialGidSet[NGROUPS_MAX];
#endif /* NGROUPS_MAX */
if (stat(qd, &st) < 0)
{
syserr("Cannot stat %s",
qid_printqueue(qgrp, qdir));
return 0;
}
#ifdef NGROUPS_MAX
n = NGROUPS_MAX;
while (--n >= 0)
{
if (InitialGidSet[n] == st.st_gid)
break;
}
if (n < 0 && RealGid != st.st_gid)
#else /* NGROUPS_MAX */
if (RealGid != st.st_gid)
#endif /* NGROUPS_MAX */
{
usrerr("510 You are not permitted to see the queue");
setstat(EX_NOPERM);
return 0;
}
}
/*
** Read and order the queue.
*/
nrequests = gatherq(qgrp, qdir, true, NULL, NULL);
(void) sortq(Queue[qgrp]->qg_maxlist);
/*
** Print the work list that we have read.
*/
/* first see if there is anything */
if (nrequests <= 0)
{
(void) sm_io_fprintf(smioout, SM_TIME_DEFAULT, "%s is empty\n",
qid_printqueue(qgrp, qdir));
return 0;
}
sm_getla(); /* get load average */
(void) sm_io_fprintf(smioout, SM_TIME_DEFAULT, "\t\t%s (%d request%s",
qid_printqueue(qgrp, qdir),
nrequests, nrequests == 1 ? "" : "s");
if (MaxQueueRun > 0 && nrequests > MaxQueueRun)
(void) sm_io_fprintf(smioout, SM_TIME_DEFAULT,
", only %d printed", MaxQueueRun);
if (Verbose)
(void) sm_io_fprintf(smioout, SM_TIME_DEFAULT,
")\n-----Q-ID----- --Size-- -Priority- ---Q-Time--- --------Sender/Recipient--------\n");
else
(void) sm_io_fprintf(smioout, SM_TIME_DEFAULT,
")\n-----Q-ID----- --Size-- -----Q-Time----- ------------Sender/Recipient-----------\n");
for (w = WorkQ; w != NULL; w = w->w_next)
{
struct stat st;
auto time_t submittime = 0;
long dfsize;
int flags = 0;
int qfver;
char quarmsg[MAXLINE];
char statmsg[MAXLINE];
char bodytype[MAXNAME + 1];
char qf[MAXPATHLEN];
if (StopRequest)
stop_sendmail();
(void) sm_io_fprintf(smioout, SM_TIME_DEFAULT, "%13s",
w->w_name + 2);
(void) sm_strlcpyn(qf, sizeof qf, 3, qd, "/", w->w_name);
f = sm_io_open(SmFtStdio, SM_TIME_DEFAULT, qf, SM_IO_RDONLY_B,
NULL);
if (f == NULL)
{
if (errno == EPERM)
(void) sm_io_fprintf(smioout, SM_TIME_DEFAULT,
" (permission denied)\n");
else if (errno == ENOENT)
(void) sm_io_fprintf(smioout, SM_TIME_DEFAULT,
" (job completed)\n");
else
(void) sm_io_fprintf(smioout, SM_TIME_DEFAULT,
" (%s)\n",
sm_errstring(errno));
errno = 0;
continue;
}
w->w_name[0] = DATAFL_LETTER;
(void) sm_strlcpyn(qf, sizeof qf, 3, qddf, "/", w->w_name);
if (stat(qf, &st) >= 0)
dfsize = st.st_size;
else
{
ENVELOPE e;
/*
** Maybe the df file can't be statted because
** it is in a different directory than the qf file.
** In order to find out, we must read the qf file.
*/
newenvelope(&e, &BlankEnvelope, sm_rpool_new_x(NULL));
e.e_id = w->w_name + 2;
e.e_qgrp = qgrp;
e.e_qdir = qdir;
dfsize = -1;
if (readqf(&e, false))
{
char *df = queuename(&e, DATAFL_LETTER);
if (stat(df, &st) >= 0)
dfsize = st.st_size;
}
if (e.e_lockfp != NULL)
{
(void) sm_io_close(e.e_lockfp, SM_TIME_DEFAULT);
e.e_lockfp = NULL;
}
clearenvelope(&e, false, e.e_rpool);
sm_rpool_free(e.e_rpool);
}
if (w->w_lock)
(void) sm_io_fprintf(smioout, SM_TIME_DEFAULT, "*");
else if (QueueMode == QM_LOST)
(void) sm_io_fprintf(smioout, SM_TIME_DEFAULT, "?");
else if (w->w_tooyoung)
(void) sm_io_fprintf(smioout, SM_TIME_DEFAULT, "-");
else if (shouldqueue(w->w_pri, w->w_ctime))
(void) sm_io_fprintf(smioout, SM_TIME_DEFAULT, "X");
else
(void) sm_io_fprintf(smioout, SM_TIME_DEFAULT, " ");
errno = 0;
quarmsg[0] = '\0';
statmsg[0] = bodytype[0] = '\0';
qfver = 0;
while (sm_io_fgets(f, SM_TIME_DEFAULT, buf, sizeof buf) != NULL)
{
register int i;
register char *p;
if (StopRequest)
stop_sendmail();
fixcrlf(buf, true);
switch (buf[0])
{
case 'V': /* queue file version */
qfver = atoi(&buf[1]);
break;
case 'M': /* error message */
if ((i = strlen(&buf[1])) >= sizeof statmsg)
i = sizeof statmsg - 1;
memmove(statmsg, &buf[1], i);
statmsg[i] = '\0';
break;
case 'q': /* quarantine reason */
if ((i = strlen(&buf[1])) >= sizeof quarmsg)
i = sizeof quarmsg - 1;
memmove(quarmsg, &buf[1], i);
quarmsg[i] = '\0';
break;
case 'B': /* body type */
if ((i = strlen(&buf[1])) >= sizeof bodytype)
i = sizeof bodytype - 1;
memmove(bodytype, &buf[1], i);
bodytype[i] = '\0';
break;
case 'S': /* sender name */
if (Verbose)
{
(void) sm_io_fprintf(smioout,
SM_TIME_DEFAULT,
"%8ld %10ld%c%.12s ",
dfsize,
w->w_pri,
bitset(EF_WARNING, flags)
? '+' : ' ',
ctime(&submittime) + 4);
prtstr(&buf[1], 78);
}
else
{
(void) sm_io_fprintf(smioout,
SM_TIME_DEFAULT,
"%8ld %.16s ",
dfsize,
ctime(&submittime));
prtstr(&buf[1], 39);
}
if (quarmsg[0] != '\0')
{
(void) sm_io_fprintf(smioout,
SM_TIME_DEFAULT,
"\n QUARANTINE: %.*s",
Verbose ? 100 : 60,
quarmsg);
quarmsg[0] = '\0';
}
if (statmsg[0] != '\0' || bodytype[0] != '\0')
{
(void) sm_io_fprintf(smioout,
SM_TIME_DEFAULT,
"\n %10.10s",
bodytype);
if (statmsg[0] != '\0')
(void) sm_io_fprintf(smioout,
SM_TIME_DEFAULT,
" (%.*s)",
Verbose ? 100 : 60,
statmsg);
statmsg[0] = '\0';
}
break;
case 'C': /* controlling user */
if (Verbose)
(void) sm_io_fprintf(smioout,
SM_TIME_DEFAULT,
"\n\t\t\t\t\t\t(---%.64s---)",
&buf[1]);
break;
case 'R': /* recipient name */
p = &buf[1];
if (qfver >= 1)
{
p = strchr(p, ':');
if (p == NULL)
break;
p++;
}
if (Verbose)
{
(void) sm_io_fprintf(smioout,
SM_TIME_DEFAULT,
"\n\t\t\t\t\t\t");
prtstr(p, 71);
}
else
{
(void) sm_io_fprintf(smioout,
SM_TIME_DEFAULT,
"\n\t\t\t\t\t ");
prtstr(p, 38);
}
if (Verbose && statmsg[0] != '\0')
{
(void) sm_io_fprintf(smioout,
SM_TIME_DEFAULT,
"\n\t\t (%.100s)",
statmsg);
statmsg[0] = '\0';
}
break;
case 'T': /* creation time */
submittime = atol(&buf[1]);
break;
case 'F': /* flag bits */
for (p = &buf[1]; *p != '\0'; p++)
{
switch (*p)
{
case 'w':
flags |= EF_WARNING;
break;
}
}
}
}
if (submittime == (time_t) 0)
(void) sm_io_fprintf(smioout, SM_TIME_DEFAULT,
" (no control file)");
(void) sm_io_fprintf(smioout, SM_TIME_DEFAULT, "\n");
(void) sm_io_close(f, SM_TIME_DEFAULT);
}
return nrequests;
}
/*
** QUEUE_LETTER -- get the proper queue letter for the current QueueMode.
**
** Parameters:
** e -- envelope to build it in/from.
** type -- the file type, used as the first character
** of the file name.
**
** Returns:
** the letter to use
*/
static char
queue_letter(e, type)
ENVELOPE *e;
int type;
{
/* Change type according to QueueMode */
if (type == ANYQFL_LETTER)
{
if (e->e_quarmsg != NULL)
type = QUARQF_LETTER;
else
{
switch (QueueMode)
{
case QM_NORMAL:
type = NORMQF_LETTER;
break;
case QM_QUARANTINE:
type = QUARQF_LETTER;
break;
case QM_LOST:
type = LOSEQF_LETTER;
break;
default:
/* should never happen */
abort();
/* NOTREACHED */
}
}
}
return type;
}
/*
** QUEUENAME -- build a file name in the queue directory for this envelope.
**
** Parameters:
** e -- envelope to build it in/from.
** type -- the file type, used as the first character
** of the file name.
**
** Returns:
** a pointer to the queue name (in a static buffer).
**
** Side Effects:
** If no id code is already assigned, queuename() will
** assign an id code with assign_queueid(). If no queue
** directory is assigned, one will be set with setnewqueue().
*/
char *
queuename(e, type)
register ENVELOPE *e;
int type;
{