X-Git-Url: https://vcs.fsf.org/?p=exim.git;a=blobdiff_plain;f=src%2Fsrc%2Fqueue.c;h=53dc6e026c6ce6d4ceb9b364fa71c99206d83738;hp=d8c160a3af8527e01e4826289eef05f359ad240c;hb=ecf0514306dd9c0baf94c43932cf77bd25fd8df0;hpb=9f01e50d7efc5c625614e4e055790ca4a92a52a8 diff --git a/src/src/queue.c b/src/src/queue.c index d8c160a3a..53dc6e026 100644 --- a/src/src/queue.c +++ b/src/src/queue.c @@ -110,13 +110,14 @@ Arguments: 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; @@ -134,7 +135,9 @@ according to the bits of the flags variable. Get a collection of bits from the 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++) @@ -204,60 +207,63 @@ for (; i <= *subcount; i++) if (len == SPOOL_NAME_LENGTH && Ustrcmp(name + SPOOL_NAME_LENGTH - 2, "-H") == 0) - { - queue_filename *next = - store_get(sizeof(queue_filename) + Ustrlen(name), is_tainted(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), is_tainted(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 */ @@ -294,7 +300,7 @@ for (; i <= *subcount; i++) /* 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]); @@ -346,6 +352,7 @@ const pcre *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 */ #ifdef MEASURE_TIMING report_time_since(×tamp_startup, US"queue_run start"); @@ -441,7 +448,7 @@ for (int i = queue_run_in_order ? -1 : 0; } 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; @@ -468,18 +475,41 @@ for (int i = queue_run_in_order ? -1 : 0; (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++; + DEBUG(D_queue_run) debug_printf("q2stage forking\n"); + if ((qpid[i] = fork())) + continue; /* parent loops around */ + DEBUG(D_queue_run) debug_printf("q2stage child\n"); + } + /* 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 @@ -499,7 +529,7 @@ for (int i = queue_run_in_order ? -1 : 0; 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 @@ -571,7 +601,7 @@ for (int i = queue_run_in_order ? -1 : 0; 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 @@ -638,11 +668,11 @@ for (int i = queue_run_in_order ? -1 : 0; /* 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); @@ -654,19 +684,32 @@ for (int i = queue_run_in_order ? -1 : 0; 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, US"2-phase child"); + /* 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, US"2-phase child"); } /* End loop for list of messages */ tree_nonrecipients = NULL; @@ -694,6 +737,19 @@ turned off. */ 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); } @@ -718,26 +774,38 @@ if (!recurse) /* 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 * @@ -819,7 +887,8 @@ else -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; @@ -1265,6 +1334,14 @@ switch(action) } + 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); @@ -1412,34 +1489,69 @@ queue_check_only(void) { 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); + } +} + + + +/******************************************************************************/ +/******************************************************************************/ + +#ifdef EXPERIMENTAL_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}; + int slen; + +#ifdef EXIM_HAVE_ABSTRACT_UNIX_SOCKETS + int len = offsetof(struct sockaddr_un, sun_path) + 1 + + snprintf(sa_un.sun_path+1, sizeof(sa_un.sun_path)-1, "%s", + NOTIFIER_SOCKET_NAME); + sa_un.sun_path[0] = 0; +#else + int len = offsetof(struct sockaddr_un, sun_path) + + snprintf(sa_un.sun_path, sizeof(sa_un.sun_path), "%s/%s", + spool_directory, NOTIFIER_SOCKET_NAME); +#endif + + if (sendto(fd, buf, sizeof(buf), 0, (struct sockaddr *)&sa_un, 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*/