* Exim - an Internet mail transport agent *
*************************************************/
+/* Copyright (c) The Exim Maintainers 2020 - 2022 */
/* Copyright (c) University of Cambridge 1995 - 2018 */
/* See the file NOTICE for conditions of use and distribution. */
+/* SPDX-License-Identifier: GPL-2.0-only */
/* Functions that operate on the input queue. */
-/* Routines with knowledge of spool layout */
-
-#ifndef COMPILE_UTILITY
-static void
-spool_pname_buf(uschar * buf, int len)
-{
-snprintf(CS buf, len, "%s/%s/input", spool_directory, queue_name);
-}
-
-uschar *
-spool_dname(const uschar * purpose, uschar * subdir)
-{
-return string_sprintf("%s/%s/%s/%s",
- spool_directory, queue_name, purpose, subdir);
-}
-#endif
-
-uschar *
-spool_sname(const uschar * purpose, uschar * subdir)
-{
-return string_sprintf("%s%s%s%s%s",
- queue_name, *queue_name ? "/" : "",
- purpose,
- *subdir ? "/" : "", subdir);
-}
-
-uschar *
-spool_fname(const uschar * purpose, const uschar * subdir, const uschar * fname,
- const uschar * suffix)
-{
-return string_sprintf("%s/%s/%s/%s/%s%s",
- spool_directory, queue_name, purpose, subdir, fname, suffix);
-}
#define LOG2_MAXNODES 32
+#ifndef DISABLE_TLS
+static BOOL queue_tls_init = FALSE;
+#endif
/*************************************************
* Helper sort function for queue_get_spool_list *
subdirs vector to store list of subdirchars
subcount pointer to int in which to store count of subdirs
randomize TRUE if the order of the list is to be unpredictable
+ pcount If not NULL, fill in with count of files and do not return list
Returns: pointer to a chain of queue name items
*/
static queue_filename *
queue_get_spool_list(int subdiroffset, uschar *subdirs, int *subcount,
- BOOL randomize)
+ BOOL randomize, unsigned * pcount)
{
int i;
int flags = 0;
int subptr;
queue_filename *yield = NULL;
queue_filename *last = NULL;
-struct dirent *ent;
-DIR *dd;
uschar buffer[256];
queue_filename *root[LOG2_MAXNODES];
current time. Use the bottom 16 and just keep re-using them if necessary. When
not randomizing, initialize the sublists for the bottom-up merge sort. */
-if (randomize)
+if (pcount)
+ *pcount = 0;
+else if (randomize)
resetflags = time(NULL) & 0xFFFF;
else
for (i = 0; i < LOG2_MAXNODES; i++)
{
int count = 0;
int subdirchar = subdirs[i]; /* 0 for main directory */
+ DIR *dd;
if (subdirchar != 0)
{
}
DEBUG(D_queue_run) debug_printf("looking in %s\n", buffer);
- if (!(dd = opendir(CS buffer)))
+ if (!(dd = exim_opendir(buffer)))
continue;
/* Now scan the directory. */
- while ((ent = readdir(dd)))
+ for (struct dirent *ent; ent = readdir(dd); )
{
uschar *name = US ent->d_name;
int len = Ustrlen(name);
if (len == SPOOL_NAME_LENGTH &&
Ustrcmp(name + SPOOL_NAME_LENGTH - 2, "-H") == 0)
- {
- queue_filename *next =
- store_get(sizeof(queue_filename) + Ustrlen(name));
- Ustrcpy(next->text, name);
- next->dir_uschar = subdirchar;
-
- /* Handle the creation of a randomized list. The first item becomes both
- the top and bottom of the list. Subsequent items are inserted either at
- the top or the bottom, randomly. This is, I argue, faster than doing a
- sort by allocating a random number to each item, and it also saves having
- to store the number with each item. */
-
- if (randomize)
- if (!yield)
- {
- next->next = NULL;
- yield = last = next;
- }
- else
- {
- if (flags == 0)
- flags = resetflags;
- if ((flags & 1) == 0)
- {
- next->next = yield;
- yield = next;
- }
- else
- {
- next->next = NULL;
- last->next = next;
- last = next;
- }
- flags = flags >> 1;
- }
+ if (pcount)
+ (*pcount)++;
+ else
+ {
+ queue_filename * next =
+ store_get(sizeof(queue_filename) + Ustrlen(name), name);
+ Ustrcpy(next->text, name);
+ next->dir_uschar = subdirchar;
+
+ /* Handle the creation of a randomized list. The first item becomes both
+ the top and bottom of the list. Subsequent items are inserted either at
+ the top or the bottom, randomly. This is, I argue, faster than doing a
+ sort by allocating a random number to each item, and it also saves having
+ to store the number with each item. */
+
+ if (randomize)
+ if (!yield)
+ {
+ next->next = NULL;
+ yield = last = next;
+ }
+ else
+ {
+ if (flags == 0)
+ flags = resetflags;
+ if ((flags & 1) == 0)
+ {
+ next->next = yield;
+ yield = next;
+ }
+ else
+ {
+ next->next = NULL;
+ last->next = next;
+ last = next;
+ }
+ flags = flags >> 1;
+ }
- /* Otherwise do a bottom-up merge sort based on the name. */
+ /* Otherwise do a bottom-up merge sort based on the name. */
- else
- {
- next->next = NULL;
- for (int j = 0; j < LOG2_MAXNODES; j++)
- if (root[j])
- {
- next = merge_queue_lists(next, root[j]);
- root[j] = j == LOG2_MAXNODES - 1 ? next : NULL;
- }
- else
- {
- root[j] = next;
- break;
- }
- }
- }
+ else
+ {
+ next->next = NULL;
+ for (int j = 0; j < LOG2_MAXNODES; j++)
+ if (root[j])
+ {
+ next = merge_queue_lists(next, root[j]);
+ root[j] = j == LOG2_MAXNODES - 1 ? next : NULL;
+ }
+ else
+ {
+ root[j] = next;
+ break;
+ }
+ }
+ }
}
/* Finished with this directory */
/* When using a bottom-up merge sort, do the final merging of the sublists.
Then pass back the final list of file items. */
-if (!randomize)
+if (!pcount && !randomize)
for (i = 0; i < LOG2_MAXNODES; ++i)
yield = merge_queue_lists(yield, root[i]);
{
BOOL force_delivery = f.queue_run_force || deliver_selectstring != NULL ||
deliver_selectstring_sender != NULL;
-const pcre *selectstring_regex = NULL;
-const pcre *selectstring_regex_sender = NULL;
+const pcre2_code *selectstring_regex = NULL;
+const pcre2_code *selectstring_regex_sender = NULL;
uschar *log_detail = NULL;
int subcount = 0;
uschar subdirs[64];
+pid_t qpid[4] = {0}; /* Parallelism factor for q2stage 1st phase */
+BOOL single_id = FALSE;
+
+#ifdef MEASURE_TIMING
+report_time_since(×tamp_startup, US"queue_run start");
+#endif
/* Cancel any specific queue domains. Turn off the flag that causes SMTP
deliveries not to happen, unless doing a 2-stage queue run, when the SMTP flag
p += sprintf(CS p, " -q%s", extras);
if (deliver_selectstring)
- p += sprintf(CS p, " -R%s %s", f.deliver_selectstring_regex? "r" : "",
- deliver_selectstring);
+ {
+ snprintf(CS p, big_buffer_size - (p - big_buffer), " -R%s %s",
+ f.deliver_selectstring_regex? "r" : "", deliver_selectstring);
+ p += Ustrlen(CCS p);
+ }
if (deliver_selectstring_sender)
- p += sprintf(CS p, " -S%s %s", f.deliver_selectstring_sender_regex? "r" : "",
- deliver_selectstring_sender);
+ {
+ snprintf(CS p, big_buffer_size - (p - big_buffer), " -S%s %s",
+ f.deliver_selectstring_sender_regex? "r" : "", deliver_selectstring_sender);
+ p += Ustrlen(CCS p);
+ }
log_detail = string_copy(big_buffer);
if (*queue_name)
queue_name, log_detail);
else
log_write(L_queue_run, LOG_MAIN, "Start queue run: %s", log_detail);
+
+ single_id = start_id && stop_id && !f.queue_2stage
+ && Ustrcmp(start_id, stop_id) == 0;
}
/* If deliver_selectstring is a regex, compile it. */
if (deliver_selectstring && f.deliver_selectstring_regex)
- selectstring_regex = regex_must_compile(deliver_selectstring, TRUE, FALSE);
+ selectstring_regex = regex_must_compile(deliver_selectstring, MCS_CASELESS, FALSE);
if (deliver_selectstring_sender && f.deliver_selectstring_sender_regex)
selectstring_regex_sender =
- regex_must_compile(deliver_selectstring_sender, TRUE, FALSE);
+ regex_must_compile(deliver_selectstring_sender, MCS_CASELESS, FALSE);
/* If the spool is split into subdirectories, we want to process it one
directory at a time, so as to spread out the directory scanning and the
i <= (queue_run_in_order ? -1 : subcount);
i++)
{
- void *reset_point1 = store_get(0);
+ rmark reset_point1 = store_mark();
DEBUG(D_queue_run)
{
}
for (queue_filename * fq = queue_get_spool_list(i, subdirs, &subcount,
- !queue_run_in_order);
+ !queue_run_in_order, NULL);
fq; fq = fq->next)
{
pid_t pid;
(double)load_average/1000.0,
(double)deliver_queue_load_max/1000.0);
+ /* If initial of a 2-phase run, maintain a set of child procs
+ to get disk parallelism */
+
+ if (f.queue_2stage && !queue_run_in_order)
+ {
+ int i;
+ if (qpid[f.running_in_test_harness ? 0 : nelem(qpid) - 1])
+ {
+ DEBUG(D_queue_run) debug_printf("q2stage waiting for child %d\n", (int)qpid[0]);
+ waitpid(qpid[0], NULL, 0);
+ DEBUG(D_queue_run) debug_printf("q2stage reaped child %d\n", (int)qpid[0]);
+ if (f.running_in_test_harness) i = 0;
+ else for (i = 0; i < nelem(qpid) - 1; i++) qpid[i] = qpid[i+1];
+ qpid[i] = 0;
+ }
+ else
+ for (i = 0; qpid[i]; ) i++;
+ if ((qpid[i] = exim_fork(US"qrun-phase-one")))
+ continue; /* parent loops around */
+ }
+
/* Skip this message unless it's within the ID limits */
if (stop_id && Ustrncmp(fq->text, stop_id, MESSAGE_ID_LENGTH) > 0)
- continue;
+ goto go_around;
if (start_id && Ustrncmp(fq->text, start_id, MESSAGE_ID_LENGTH) < 0)
- continue;
+ goto go_around;
/* Check that the message still exists */
message_subdir[0] = fq->dir_uschar;
if (Ustat(spool_fname(US"input", message_subdir, fq->text, US""), &statbuf) < 0)
- continue;
+ goto go_around;
/* There are some tests that require the reading of the header file. Ensure
the store used is scavenged afterwards so that this process doesn't keep
{
BOOL wanted = TRUE;
BOOL orig_dont_deliver = f.dont_deliver;
- void *reset_point2 = store_get(0);
+ rmark reset_point2 = store_mark();
/* Restore the original setting of dont_deliver after reading the header,
so that a setting for a particular message doesn't force it for any that
follow. If the message is chosen for delivery, the header is read again
in the deliver_message() function, in a subprocess. */
- if (spool_read_header(fq->text, FALSE, TRUE) != spool_read_OK) continue;
+ if (spool_read_header(fq->text, FALSE, TRUE) != spool_read_OK) goto go_around;
f.dont_deliver = orig_dont_deliver;
/* Now decide if we want to deliver this message. As we have read the
else if ( deliver_selectstring_sender
&& !(f.deliver_selectstring_sender_regex
- ? (pcre_exec(selectstring_regex_sender, NULL,
- CS sender_address, Ustrlen(sender_address), 0, PCRE_EOPT,
- NULL, 0) >= 0)
+ ? regex_match(selectstring_regex_sender, sender_address, -1, NULL)
: (strstric(sender_address, deliver_selectstring_sender, FALSE)
!= NULL)
) )
{
uschar *address = recipients_list[i].address;
if ( (f.deliver_selectstring_regex
- ? (pcre_exec(selectstring_regex, NULL, CS address,
- Ustrlen(address), 0, PCRE_EOPT, NULL, 0) >= 0)
+ ? regex_match(selectstring_regex, address, -1, NULL)
: (strstric(address, deliver_selectstring, FALSE) != NULL)
)
&& tree_search(tree_nonrecipients, address) == NULL
spool_clear_header_globals();
store_reset(reset_point2);
- if (!wanted) continue; /* With next message */
+ if (!wanted) goto go_around; /* With next message */
}
/* OK, got a message we want to deliver. Create a pipe which will
set_process_info("running queue: %s", fq->text);
fq->text[SPOOL_NAME_LENGTH-2] = 0;
- if ((pid = fork()) == 0)
+#ifdef MEASURE_TIMING
+ report_time_since(×tamp_startup, US"queue msg selected");
+#endif
+
+#ifndef DISABLE_TLS
+ if (!queue_tls_init)
+ {
+ queue_tls_init = TRUE;
+ /* Preload TLS library info for smtp transports. Once, and only if we
+ have a delivery to do. */
+ tls_client_creds_reload(FALSE);
+ }
+#endif
+
+single_item_retry:
+ if ((pid = exim_fork(US"qrun-delivery")) == 0)
{
int rc;
- if (f.running_in_test_harness) millisleep(100);
(void)close(pfd[pipe_read]);
rc = deliver_message(fq->text, force_delivery, FALSE);
- _exit(rc == DELIVER_NOT_ATTEMPTED);
+ exim_underbar_exit(rc == DELIVER_NOT_ATTEMPTED
+ ? EXIT_FAILURE : EXIT_SUCCESS);
}
if (pid < 0)
log_write(0, LOG_MAIN|LOG_PANIC_DIE, "fork of delivery process from "
/* A zero return means a delivery was attempted; turn off the force flag
for any subsequent calls unless queue_force is set. */
- if ((status & 0xffff) == 0) force_delivery = f.queue_run_force;
+ if (!(status & 0xffff)) force_delivery = f.queue_run_force;
/* If the process crashed, tell somebody */
- else if ((status & 0x00ff) != 0)
+ else if (status & 0x00ff)
log_write(0, LOG_MAIN|LOG_PANIC,
"queue run: process %d crashed with signal %d while delivering %s",
(int)pid, status & 0x00ff, fq->text);
+ /* If single-item delivery was untried (likely due to locking)
+ retry once after a delay */
+
+ if (status & 0xff00 && single_id)
+ {
+ single_id = FALSE;
+ DEBUG(D_queue_run) debug_printf("qrun single-item pause before retry\n");
+ millisleep(500);
+ DEBUG(D_queue_run) debug_printf("qrun single-item retry after pause\n");
+ goto single_item_retry;
+ }
+
/* Before continuing, wait till the pipe gets closed at the far end. This
tells us that any children created by the delivery to re-use any SMTP
channels have all finished. Since no process actually writes to the pipe,
set_process_info("running queue: waiting for children of %d", pid);
if ((status = read(pfd[pipe_read], buffer, sizeof(buffer))) != 0)
- log_write(0, LOG_MAIN|LOG_PANIC, "queue run: %s on pipe",
- status > 0 ? "unexpected data" : "error");
+ log_write(0, LOG_MAIN|LOG_PANIC, status > 0 ?
+ "queue run: unexpected data on pipe" : "queue run: error on pipe: %s",
+ strerror(errno));
(void)close(pfd[pipe_read]);
set_process_info("running queue");
+ /* If initial of a 2-phase run, we are a child - so just exit */
+ if (f.queue_2stage && !queue_run_in_order)
+ exim_exit(EXIT_SUCCESS);
+
/* If we are in the test harness, and this is not the first of a 2-stage
queue run, update fudged queue times. */
if (f.running_in_test_harness && !f.queue_2stage)
{
- uschar *fqtnext = Ustrchr(fudged_queue_times, '/');
- if (fqtnext != NULL) fudged_queue_times = fqtnext + 1;
+ uschar * fqtnext = Ustrchr(fudged_queue_times, '/');
+ if (fqtnext) fudged_queue_times = fqtnext + 1;
}
+
+
+ continue;
+
+ go_around:
+ /* If initial of a 2-phase run, we are a child - so just exit */
+ if (f.queue_2stage && !queue_run_in_order)
+ exim_exit(EXIT_SUCCESS);
} /* End loop for list of messages */
tree_nonrecipients = NULL;
if (f.queue_2stage)
{
+
+ /* wait for last children */
+ for (int i = 0; i < nelem(qpid); i++)
+ if (qpid[i])
+ {
+ DEBUG(D_queue_run) debug_printf("q2stage reaped child %d\n", (int)qpid[i]);
+ waitpid(qpid[i], NULL, 0);
+ }
+ else break;
+
+#ifdef MEASURE_TIMING
+ report_time_since(×tamp_startup, US"queue_run 1st phase done");
+#endif
f.queue_2stage = FALSE;
queue_run(start_id, stop_id, TRUE);
}
/* Called as a result of -bpc
Arguments: none
-Returns: nothing
+Returns: count
*/
-void
+unsigned
queue_count(void)
{
int subcount;
-int count = 0;
+unsigned count = 0;
uschar subdirs[64];
-for (queue_filename *f = queue_get_spool_list(
- -1, /* entire queue */
- subdirs, /* for holding sub list */
- &subcount, /* for subcount */
- FALSE); /* not random */
- f; f = f->next) count++;
-fprintf(stdout, "%d\n", count);
+(void) queue_get_spool_list(-1, /* entire queue */
+ subdirs, /* for holding sub list */
+ &subcount, /* for subcount */
+ FALSE, /* not random */
+ &count); /* just get the count */
+return count;
}
+#define QUEUE_SIZE_AGE 60 /* update rate for queue_size */
+
+unsigned
+queue_count_cached(void)
+{
+time_t now;
+if ((now = time(NULL)) >= queue_size_next)
+ {
+ queue_size = queue_count();
+ queue_size_next = now + (f.running_in_test_harness ? 3 : QUEUE_SIZE_AGE);
+ }
+return queue_size;
+}
/************************************************
* List extra deliveries *
{
int subcount;
int now = (int)time(NULL);
-void *reset_point;
+rmark reset_point;
queue_filename * qf = NULL;
uschar subdirs[64];
queue_filename *last = NULL;
for (int i = 0; i < count; i++)
{
- queue_filename *next =
- store_get(sizeof(queue_filename) + Ustrlen(list[i]) + 2);
+ queue_filename * next =
+ store_get(sizeof(queue_filename) + Ustrlen(list[i]) + 2, list[i]);
sprintf(CS next->text, "%s-H", list[i]);
next->dir_uschar = '*';
next->next = NULL;
-1, /* entire queue */
subdirs, /* for holding sub list */
&subcount, /* for subcount */
- option >= 8); /* randomize if required */
+ option >= 8, /* randomize if required */
+ NULL); /* don't just count */
if (option >= 8) option -= 8;
/* Now scan the chain and print information, resetting store used
each time. */
-for (reset_point = store_get(0);
- qf;
+for (;
+ qf && (reset_point = store_mark());
spool_clear_header_globals(), store_reset(reset_point), qf = qf->next
)
{
for (int i = 0; i < 2; i++)
{
- message_subdir[0] = split_spool_directory == (i == 0) ? id[5] : 0;
+ set_subdir_str(message_subdir, id, i);
if ((fd = Uopen(spool_fname(subdirectory, message_subdir, id, suffix),
O_RDONLY, 0)) >= 0)
break;
if (removed)
{
#ifndef DISABLE_EVENT
- for (int i = 0; i < recipients_count; i++)
+ if (event_action) for (int i = 0; i < recipients_count; i++)
{
tree_node *delivered =
tree_search(tree_nonrecipients, recipients_list[i].address);
deliver_domain = dom
? CUS string_copyn(addr+dom, end - dom) : CUS"";
- event_raise(event_action, US"msg:fail:internal",
- string_sprintf("message removed by %s", username));
+ (void) event_raise(event_action, US"msg:fail:internal",
+ string_sprintf("message removed by %s", username), NULL);
deliver_localpart = save_local;
deliver_domain = save_domain;
}
}
}
- (void) event_raise(event_action, US"msg:complete", NULL);
+ (void) event_raise(event_action, US"msg:complete", NULL, NULL);
#endif
log_write(0, LOG_MAIN, "removed by %s", username);
log_write(0, LOG_MAIN, "Completed");
}
+ case MSG_SETQUEUE:
+ /* The global "queue_name_dest" is used as destination, "queue_name"
+ as source */
+
+ spool_move_message(id, message_subdir, US"", US"");
+ break;
+
+
case MSG_MARK_ALL_DELIVERED:
for (int i = 0; i < recipients_count; i++)
tree_add_nonrecipient(recipients_list[i].address);
parse_extract_address(argv[recipients_arg], &errmess, &start, &end,
&domain, (action == MSG_EDIT_SENDER));
- if (recipient == NULL)
+ if (!recipient)
{
yield = FALSE;
printf("- error while %s:\n bad address %s: %s\n",
doing, argv[recipients_arg], errmess);
}
- else if (recipient[0] != 0 && domain == 0)
+ else if (*recipient && domain == 0)
{
yield = FALSE;
printf("- error while %s:\n bad address %s: "
{
int sep = 0;
struct stat statbuf;
-const uschar *s;
-uschar *ss;
-uschar buffer[1024];
-
-if (queue_only_file == NULL) return;
+const uschar * s = queue_only_file;
+uschar * ss;
-s = queue_only_file;
-while ((ss = string_nextinlist(&s, &sep, buffer, sizeof(buffer))) != NULL)
- {
- if (Ustrncmp(ss, "smtp", 4) == 0)
- {
- ss += 4;
- if (Ustat(ss, &statbuf) == 0)
- {
- f.queue_smtp = TRUE;
- DEBUG(D_receive) debug_printf("queue_smtp set because %s exists\n", ss);
- }
- }
- else
- {
- if (Ustat(ss, &statbuf) == 0)
+if (s)
+ while ((ss = string_nextinlist(&s, &sep, NULL, 0)))
+ if (Ustrncmp(ss, "smtp", 4) == 0)
{
- queue_only = TRUE;
- DEBUG(D_receive) debug_printf("queue_only set because %s exists\n", ss);
+ ss += 4;
+ if (Ustat(ss, &statbuf) == 0)
+ {
+ f.queue_smtp = TRUE;
+ DEBUG(D_receive) debug_printf("queue_smtp set because %s exists\n", ss);
+ }
}
- }
+ else
+ if (Ustat(ss, &statbuf) == 0)
+ {
+ queue_only = TRUE;
+ DEBUG(D_receive) debug_printf("queue_only set because %s exists\n", ss);
+ }
+}
+
+
+
+/******************************************************************************/
+/******************************************************************************/
+
+#ifndef DISABLE_QUEUE_RAMP
+void
+queue_notify_daemon(const uschar * msgid)
+{
+uschar buf[MESSAGE_ID_LENGTH + 2];
+int fd;
+
+DEBUG(D_queue_run) debug_printf("%s: %s\n", __FUNCTION__, msgid);
+
+buf[0] = NOTIFY_MSG_QRUN;
+memcpy(buf+1, msgid, MESSAGE_ID_LENGTH+1);
+
+if ((fd = socket(AF_UNIX, SOCK_DGRAM, 0)) >= 0)
+ {
+ struct sockaddr_un sa_un = {.sun_family = AF_UNIX};
+ ssize_t len = daemon_notifier_sockname(&sa_un);
+
+ if (sendto(fd, buf, sizeof(buf), 0, (struct sockaddr *)&sa_un, (socklen_t)len) < 0)
+ DEBUG(D_queue_run)
+ debug_printf("%s: sendto %s\n", __FUNCTION__, strerror(errno));
+ close(fd);
}
+else DEBUG(D_queue_run) debug_printf(" socket: %s\n", strerror(errno));
}
+#endif
#endif /*!COMPILE_UTILITY*/
git://git.exim.org / exim.git / blobdiff