US strerror(addr->basic_errno));
if (addr->host_used)
+ {
s = string_append(s, &size, &ptr, 5,
US" H=", addr->host_used->name,
US" [", addr->host_used->address, US"]");
+ if (LOGGING(outgoing_port))
+ {
+ int port = addr->host_used->port;
+ s = string_append(s, &size, &ptr, 2,
+ US":", port == PORT_NONE ? US"25" : string_sprintf("%d", port));
+ }
+ }
if (addr->message)
s = string_append(s, &size, &ptr, 2, US": ", addr->message);
if ( !shadowing
&& ( tp->return_output || tp->return_fail_output
- || tp->log_output || tp->log_fail_output
+ || tp->log_output || tp->log_fail_output || tp->log_defer_output
) )
{
uschar *error;
}
}
-/*XXX prefer to do max_parallel check before we fork. Are we allowed to defer
-this late (we could be a shadow tpt)? */
-
/* Create the pipe for inter-process communication. */
if (pipe(pfd) != 0)
+
+/* Check transport for the given concurrency limit. Return TRUE if over
+the limit (or an expansion failure), else FALSE and if there was a limit,
+the key for the hints database used for the concurrency count. */
+
+static BOOL
+tpt_parallel_check(transport_instance * tp, address_item * addr, uschar ** key)
+{
+unsigned max_parallel;
+
+if (!tp->max_parallel) return FALSE;
+
+max_parallel = (unsigned) expand_string_integer(tp->max_parallel, TRUE);
+if (expand_string_message)
+ {
+ log_write(0, LOG_MAIN|LOG_PANIC, "Failed to expand max_parallel option "
+ "in %s transport (%s): %s", tp->name, addr->address,
+ expand_string_message);
+ return TRUE;
+ }
+
+if (max_parallel > 0)
+ {
+ uschar * serialize_key = string_sprintf("tpt-serialize-%s", tp->name);
+ if (!enq_start(serialize_key, max_parallel))
+ {
+ address_item * next;
+ DEBUG(D_transport)
+ debug_printf("skipping tpt %s because concurrency limit %u reached\n",
+ tp->name, max_parallel);
+ do
+ {
+ next = addr->next;
+ addr->message = US"concurrency limit reached for transport";
+ addr->basic_errno = ERRNO_TRETRY;
+ post_process_one(addr, DEFER, LOG_MAIN, DTYPE_TRANSPORT, 0);
+ } while ((addr = next));
+ return TRUE;
+ }
+ *key = serialize_key;
+ }
+return FALSE;
+}
+
+
+
/*************************************************
* Do local deliveries *
*************************************************/
int logflags = LOG_MAIN;
int logchar = dont_deliver? '*' : '=';
transport_instance *tp;
+ uschar * serialize_key = NULL;
/* Pick the first undelivered address off the chain */
last = next;
batch_count++;
}
- else anchor = &(next->next); /* Skip the address */
+ else anchor = &next->next; /* Skip the address */
}
}
if (!addr) continue;
+ /* If the transport is limited for parallellism, enforce that here.
+ We use a hints DB entry, incremented here and decremented after
+ the transport (and any shadow transport) completes. */
+
+ if (tpt_parallel_check(tp, addr, &serialize_key))
+ {
+ if (expand_string_message)
+ {
+ logflags |= LOG_PANIC;
+ do
+ {
+ addr = addr->next;
+ post_process_one(addr, DEFER, logflags, DTYPE_TRANSPORT, 0);
+ } while ((addr = addr2));
+ }
+ continue; /* Loop for the next set of addresses. */
+ }
+
+
/* So, finally, we do have some addresses that can be passed to the
transport. Before doing so, set up variables that are relevant to a
single delivery. */
deliver_set_expansions(NULL);
+ /* If the transport was parallelism-limited, decrement the hints DB record. */
+
+ if (serialize_key) enq_end(serialize_key);
+
/* Now we can process the results of the real transport. We must take each
address off the chain first, because post_process_one() puts it on another
chain. */
/* copy and read header */
memcpy(header, ptr, PIPE_HEADER_SIZE);
- header[PIPE_HEADER_SIZE] = '\0';
+ header[PIPE_HEADER_SIZE] = '\0';
id = header[0];
subid = header[1];
required = Ustrtol(header + 2, &endc, 10) + PIPE_HEADER_SIZE; /* header + data */
}
DEBUG(D_deliver)
- debug_printf("header read id:%c,subid:%c,size:%s,required:%d,remaining:%d,unfinished:%d\n",
+ debug_printf("header read id:%c,subid:%c,size:%s,required:%d,remaining:%d,unfinished:%d\n",
id, subid, header+2, required, remaining, unfinished);
/* is there room for the dataset we want to read ? */
break;
}
- /* we wrote all datasets with atomic write() calls
+ /* we wrote all datasets with atomic write() calls
remaining < required only happens if big_buffer was too small
- to get all available data from pipe. unfinished has to be true
+ to get all available data from pipe. unfinished has to be true
as well. */
if (remaining < required)
{
if (unfinished)
continue;
msg = string_sprintf("failed to read pipe from transport process "
- "%d for transport %s: required size=%d > remaining size=%d and unfinished=false",
+ "%d for transport %s: required size=%d > remaining size=%d and unfinished=false",
pid, addr->transport->driver_name, required, remaining);
done = TRUE;
break;
/* step behind the header */
ptr += PIPE_HEADER_SIZE;
-
+
/* Handle each possible type of item, assuming the complete item is
available in store. */
"remote delivery process count got out of step");
parcount = 0;
}
- else remote_post_process(doneaddr, LOG_MAIN, NULL, fallback);
+ else
+ {
+ transport_instance * tp = doneaddr->transport;
+ if (tp->max_parallel)
+ enq_end(string_sprintf("tpt-serialize-%s", tp->name));
+
+ remote_post_process(doneaddr, LOG_MAIN, NULL, fallback);
+ }
}
}
if (size > 99999)
{
- log_write(0, LOG_MAIN|LOG_PANIC_DIE,
+ log_write(0, LOG_MAIN|LOG_PANIC_DIE,
"Failed writing transport result to pipe: can't handle buffers > 99999 bytes. truncating!\n");
size = 99999;
}
writebuffer[0] = '\0';
}
-DEBUG(D_deliver) debug_printf("header write id:%c,subid:%c,size:%d,final:%s\n",
+DEBUG(D_deliver) debug_printf("header write id:%c,subid:%c,size:%d,final:%s\n",
id, subid, size, writebuffer);
if (buf && size > 0)
address_item *last = addr;
address_item *next;
uschar * panicmsg;
+ uschar * serialize_key = NULL;
/* Pull the first address right off the list. */
return FALSE;
}
+ /* If the transport is limited for parallellism, enforce that here.
+ The hints DB entry is decremented in par_reduce(), when we reap the
+ transport process. */
+
+ if (tpt_parallel_check(tp, addr, &serialize_key))
+ if ((panicmsg = expand_string_message))
+ goto panic_continue;
+ else
+ continue; /* Loop for the next set of addresses. */
+
/* Set up the expansion variables for this set of addresses */
deliver_set_expansions(addr);
{
panicmsg = string_sprintf("Failed to expand return path \"%s\": %s",
tp->return_path, expand_string_message);
- goto panic_continue;
+ goto enq_continue;
}
}
if (!findugid(addr, tp, &uid, &gid, &use_initgroups))
{
panicmsg = NULL;
- goto panic_continue;
+ goto enq_continue;
}
/* If this transport has a setup function, call it now so that it gets
if (!ok)
{
DEBUG(D_deliver) debug_printf("not suitable for continue_transport\n");
- next = addr;
+ if (serialize_key) enq_end(serialize_key);
if (addr->fallback_hosts && !fallback)
{
- for (;; next = next->next)
+ for (next = addr; ; next = next->next)
{
next->host_list = next->fallback_hosts;
DEBUG(D_deliver) debug_printf("%s queued for fallback host(s)\n", next->address);
}
else
- {
- while (next->next) next = next->next;
- next->next = addr_defer;
- addr_defer = addr;
- }
+ {
+ while (next->next) next = next->next;
+ next->next = addr_defer;
+ addr_defer = addr;
+ }
continue;
}
if (!pipe_done)
{
panicmsg = string_sprintf("unable to create pipe: %s", strerror(errno));
- goto panic_continue;
+ goto enq_continue;
}
/* Find a free slot in the pardata list. Must do this after the possible
(void)close(pfd[pipe_write]);
(void)close(pfd[pipe_read]);
panicmsg = US"Unexpectedly no free subprocess slot";
- goto panic_continue;
+ goto enq_continue;
}
/* Now fork a subprocess to do the remote delivery, but before doing so,
(void)close(pfd[pipe_read]);
panicmsg = string_sprintf("fork failed for remote delivery to %s: %s",
addr->domain, strerror(errno));
- goto panic_continue;
+ goto enq_continue;
}
/* Fork succeeded; increment the count, and remember relevant data for
continue;
+enq_continue:
+ if (serialize_key) enq_end(serialize_key);
panic_continue:
remote_post_process(addr, LOG_MAIN|LOG_PANIC, panicmsg, fallback);
continue;
if (r->pno >= 0)
new->onetime_parent = recipients_list[r->pno].address;
- /* If DSN support is enabled, set the dsn flags and the original receipt
+ /* If DSN support is enabled, set the dsn flags and the original receipt
to be passed on to other DSN enabled MTAs */
new->dsn_flags = r->dsn_flags & rf_dsnflags;
new->dsn_orcpt = r->orcpt;
so just queue them all. */
if (queue_run_local)
- {
while (addr_remote)
{
address_item *addr = addr_remote;
addr->message = US"remote deliveries suppressed";
(void)post_process_one(addr, DEFER, LOG_MAIN, DTYPE_TRANSPORT, 0);
}
- }
/* Handle remote deliveries */
addr_senddsn->next = addr_next;
}
else
- DEBUG(D_deliver) debug_printf("DSN: not sending DSN success message\n");
+ DEBUG(D_deliver) debug_printf("DSN: not sending DSN success message\n");
addr_dsntmp = addr_dsntmp->next;
}
pid_t pid;
int fd;
- /* create exim process to send message */
+ /* create exim process to send message */
pid = child_open_exim(&fd);
DEBUG(D_deliver) debug_printf("DSN: child_open_exim returns: %d\n", pid);
-
+
if (pid < 0) /* Creation of child failed */
{
log_write(0, LOG_MAIN|LOG_PANIC_DIE, "Process %d (parent %d) failed to "
getppid(), strerror(errno));
DEBUG(D_deliver) debug_printf("DSN: child_open_exim failed\n");
- }
+ }
else /* Creation of child succeeded */
{
FILE *f = fdopen(fd, "wb");
/* header only as required by RFC. only failure DSN needs to honor RET=FULL */
int topt = topt_add_return_path | topt_no_body;
uschar * bound;
-
+
DEBUG(D_deliver)
debug_printf("sending error message to: %s\n", sender_address);
-
+
/* build unique id for MIME boundary */
bound = string_sprintf(TIME_T_FMT "-eximdsn-%d", time(NULL), rand());
DEBUG(D_deliver) debug_printf("DSN: MIME boundary: %s\n", bound);
-
+
if (errors_reply_to)
fprintf(f, "Reply-To: %s\n", errors_reply_to);
-
+
fprintf(f, "Auto-Submitted: auto-generated\n"
"From: Mail Delivery System <Mailer-Daemon@%s>\n"
"To: %s\n"
"--%s\n"
"Content-type: text/plain; charset=us-ascii\n\n"
-
+
"This message was created automatically by mail delivery software.\n"
" ----- The following addresses had successful delivery notifications -----\n",
qualify_domain_sender, sender_address, bound, bound);
}
fprintf(f, "--%s\nContent-type: text/rfc822-headers\n\n", bound);
-
+
fflush(f);
transport_filter_argv = NULL; /* Just in case */
return_path = sender_address; /* In case not previously set */
-
+
/* Write the original email out */
transport_write_message(NULL, fileno(f), topt, 0, NULL, NULL, NULL, NULL, NULL, 0);
fflush(f);
fprintf(f, "X-Original-Envelope-ID: error decoding xtext formated ENVID\n");
}
fputc('\n', f);
-
+
for (addr = handled_addr; addr; addr = addr->next)
{
host_item * hu;
if ((hu = addr->host_used) && hu->name)
{
const uschar * s;
- fprintf(f, "Remote-MTA: dns; %s\n",
- hu->name);
+ fprintf(f, "Remote-MTA: dns; %s\n", hu->name);
#ifdef EXPERIMENTAL_DSN_INFO
if (hu->address)
{
emf_text = next_emf(emf, US"copy");
/* add message body
- we ignore the intro text from template and add
+ we ignore the intro text from template and add
the text for bounce_return_size_limit at the end.
-
+
bounce_return_message is ignored
in case RET= is defined we honor these values
otherwise bounce_return_body is honored.
-
+
bounce_return_size_limit is always honored.
*/
-
+
fprintf(f, "--%s\n", bound);
dsnlimitmsg = US"X-Exim-DSN-Information: Due to administrative limits only headers are returned";
dsnnotifyhdr = dsnlimitmsg;
}
}
-
+
#ifdef EXPERIMENTAL_INTERNATIONAL
if (message_smtputf8)
fputs(topt & topt_no_body ? "Content-type: message/global-headers\n\n"
transport_write_message(NULL, fileno(f), topt,
0, dsnnotifyhdr, NULL, NULL, NULL, NULL, 0);
fflush(f);
-
+
/* we never add the final text. close the file */
if (emf)
(void)fclose(emf);
-
+
fprintf(f, "\n--%s--\n", bound);
/* Close the file, which should send an EOF to the child process
#ifdef EXPERIMENTAL_EVENT
(void) event_raise(event_action, US"msg:complete", NULL);
#endif
-}
+ }
/* If there are deferred addresses, we are keeping this message because it is
not yet completed. Lose any temporary files that were catching output from
"Reporting-MTA: dns; %s\n",
bound,
smtp_active_hostname);
-
+
if (dsn_envid)
{
addr_dsndefer->address);
if (addr_dsndefer->host_used && addr_dsndefer->host_used->name)
{
- fprintf(f, "Remote-MTA: dns; %s\n",
+ fprintf(f, "Remote-MTA: dns; %s\n",
addr_dsndefer->host_used->name);
print_dsn_diagnostic_code(addr_dsndefer, f);
}
projects / exim.git / blobdiff