queue.c

< linux网络编程工具>>配套源码

		/* make sure jobs in creation don't clog queue */
		w->w_pri = 0x7fffffff;
		w->w_ctime = 0;

		/* extract useful information */
		i = NEED_P | NEED_T;
		if (QueueLimitSender != NULL)
			i |= NEED_S;
		if (QueueLimitRecipient != NULL)
			i |= NEED_R;
		while (i != 0 && fgets(lbuf, sizeof lbuf, cf) != NULL)
		{
			int c;
			time_t age;

			p = strchr(lbuf, '\n');
			if (p != NULL)
				*p = '\0';
			else
			{
				/* flush rest of overly long line */
				while ((c = getc(cf)) != 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 'R':
				if (w->w_host == NULL &&
				    (p = strrchr(&lbuf[1], '@')) != NULL)
				{
					w->w_host = strrev(&p[1]);
					makelower(w->w_host);
				}
				if (QueueLimitRecipient == NULL)
				{
					i &= ~NEED_R;
					break;
				}
				if (qfver > 0)
				{
					p = strchr(&lbuf[1], ':');
					if (p == NULL)
						p = &lbuf[1];
				}
				else
					p = &lbuf[1];
				check = QueueLimitRecipient;
				while (check != NULL)
				{
					if (strcontainedin(check->queue_match,
							   p))
						break;
					else
						check = check->queue_next;
				}
				if (check != NULL)
					i &= ~NEED_R;
				break;

			  case 'S':
				check = QueueLimitSender;
				while (check != NULL)
				{
					if (strcontainedin(check->queue_match,
							   &lbuf[1]))
						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 _FFR_QUEUEDELAY
/*
			  case 'G':
				queuealg = atoi(lbuf[1]);
				break;
			  case 'Y':
				queuedelay = (time_t) atol(&lbuf[1]);
				break;
*/
# endif /* _FFR_QUEUEDELAY */
			}
		}
		(void) fclose(cf);

		if ((!doall && shouldqueue(w->w_pri, w->w_ctime)) ||
		    bitset(NEED_R|NEED_S, i))
		{
			/* don't even bother sorting this job in */
			if (tTd(41, 49))
				dprintf("skipping %s (%x)\n", w->w_name, i);
			free(w->w_name);
			if (w->w_host)
				free(w->w_host);
			wn--;
		}
	}
	(void) closedir(f);
	wn++;

	WorkQ = NULL;
	if (WorkList == NULL)
		return 0;
	wc = min(wn, WorkListSize);
	if (wc > MaxQueueRun && MaxQueueRun > 0)
		wc = MaxQueueRun;

	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 qf filename.
		*/

		qsort((char *) WorkList, wc, sizeof *WorkList, workcmpf4);
	}
	else
	{
		/*
		**  Simple sort based on queue priority only.
		*/

		qsort((char *) WorkList, wc, sizeof *WorkList, workcmpf0);
	}

	/*
	**  Convert the work list into canonical form.
	**	Should be turning it into a list of envelopes here perhaps.
	*/

	for (i = wc; --i >= 0; )
	{
		w = (WORK *) xalloc(sizeof *w);
		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_next = WorkQ;
		WorkQ = w;
	}
	if (WorkList != NULL)
		free(WorkList);
	WorkList = NULL;
	WorkListSize = 0;

	if (tTd(40, 1))
	{
		for (w = WorkQ; w != NULL; w = w->w_next)
		{
			if (w->w_host != NULL)
				dprintf("%22s: pri=%ld %s\n",
					w->w_name, w->w_pri, w->w_host);
			else
				dprintf("%32s: pri=%ld\n",
					w->w_name, w->w_pri);
		}
	}

	return wn;
}
/*
**  GROW_WLIST -- make the work list larger
**
**	Parameters:
**		queuedir -- 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(queuedir)
	int queuedir;
{
	if (tTd(41, 1))
		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 *) 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(queuedir),
					  WorkListSize);
			}
		}
		else if (LogLevel > 0)
		{
			sm_syslog(LOG_ALERT, NOQID,
				  "FAILED to grow WorkList for %s to %d",
				  qid_printqueue(queuedir), newsize);
		}
	}
	if (tTd(41, 1))
		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
**
**	Side Effects:
**		none.
*/

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
**
**	Side Effects:
**		none.
*/

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 a->w_pri - b->w_pri;
}
/*
**  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
**
**	Side Effects:
**		none.
*/

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 a->w_pri - b->w_pri;
}
/*
**  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
**
**	Side Effects:
**		none.
*/

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
**
**	Side Effects:
**		none.
*/

static int
workcmpf4(a, b)
	register WORK *a;
	register WORK *b;
{
	return strcmp(a->w_name, b->w_name);
}
/*
**  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;
}
/*
**  DOWORK -- do a work request.
**
**	Parameters:
**		queuedir -- 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(queuedir, id, forkflag, requeueflag, e)
	int queuedir;
	char *id;
	bool forkflag;
	bool requeueflag;
	register ENVELOPE *e;
{
	register pid_t pid;

	if (tTd(40, 1))
		dprintf("dowork(%s/%s)\n", qid_printqueue(queuedir), 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();

		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
		{
			/* 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.
		*/

		/* set basic modes, etc. */
		(void) alarm(0);
		clearstats();
		clearenvelope(e, FALSE);
		e->e_flags |= EF_QUEUERUN|EF_GLOBALERRS;
		set_delivery_mode(SM_DELIVER, e);
		e->e_errormode = EM_MAIL;
		e->e_id = id;
		e->e_queuedir = queuedir;
		GrabTo = UseErrorsTo = FALSE;
		ExitStat = EX_OK;
		if (forkflag)
		{
			disconnect(1, e);
			OpMode = MD_QUEUERUN;
		}
		sm_setproctitle(TRUE, e, "%s: from queue", qid_printname(e));
		if (LogLevel > 76)
			sm_syslog(LOG_DEBUG, e->e_id,