#include <sys_defs.h>
#include <unistd.h>
#include <sys/time.h>
#include <sys/types.h>
#include <unistd.h>
#ifdef USE_SYS_SELECT_H
#include <sys/select.h>
#endif
#include <msg.h>
#include <htable.h>
#include <events.h>
#include <mymalloc.h>
#include <vstream.h>
#include <iostuff.h>
#include <mail_proto.h>
#include <recipient_list.h>
#include <mail_conf.h>
#include <mail_params.h>
#include "qmgr.h"
HTABLE *qmgr_transport_byname;
QMGR_TRANSPORT_LIST qmgr_transport_list;
typedef struct QMGR_TRANSPORT_ALLOC QMGR_TRANSPORT_ALLOC;
struct QMGR_TRANSPORT_ALLOC {
QMGR_TRANSPORT *transport;
VSTREAM *stream;
QMGR_TRANSPORT_ALLOC_NOTIFY notify;
};
#ifndef QMGR_TRANSPORT_MAX_PEND
#define QMGR_TRANSPORT_MAX_PEND 2
#endif
static void qmgr_transport_unthrottle_wrapper(int unused_event, void *context)
{
qmgr_transport_unthrottle((QMGR_TRANSPORT *) context);
}
void qmgr_transport_unthrottle(QMGR_TRANSPORT *transport)
{
const char *myname = "qmgr_transport_unthrottle";
if ((transport->flags & QMGR_TRANSPORT_STAT_DEAD) != 0) {
if (msg_verbose)
msg_info("%s: transport %s", myname, transport->name);
transport->flags &= ~QMGR_TRANSPORT_STAT_DEAD;
if (transport->dsn == 0)
msg_panic("%s: transport %s: null reason",
myname, transport->name);
dsn_free(transport->dsn);
transport->dsn = 0;
event_cancel_timer(qmgr_transport_unthrottle_wrapper,
(void *) transport);
}
if (transport->flags & QMGR_TRANSPORT_STAT_RATE_LOCK)
transport->flags &= ~QMGR_TRANSPORT_STAT_RATE_LOCK;
}
void qmgr_transport_throttle(QMGR_TRANSPORT *transport, DSN *dsn)
{
const char *myname = "qmgr_transport_throttle";
if ((transport->flags & QMGR_TRANSPORT_STAT_DEAD) == 0) {
if (msg_verbose)
msg_info("%s: transport %s: status: %s reason: %s",
myname, transport->name, dsn->status, dsn->reason);
transport->flags |= QMGR_TRANSPORT_STAT_DEAD;
if (transport->dsn)
msg_panic("%s: transport %s: spurious reason: %s",
myname, transport->name, transport->dsn->reason);
transport->dsn = DSN_COPY(dsn);
event_request_timer(qmgr_transport_unthrottle_wrapper,
(void *) transport, var_transport_retry_time);
}
}
static void qmgr_transport_abort(int unused_event, void *context)
{
QMGR_TRANSPORT_ALLOC *alloc = (QMGR_TRANSPORT_ALLOC *) context;
msg_fatal("timeout connecting to transport: %s", alloc->transport->name);
}
static void qmgr_transport_rate_event(int unused_event, void *context)
{
QMGR_TRANSPORT_ALLOC *alloc = (QMGR_TRANSPORT_ALLOC *) context;
alloc->notify(alloc->transport, alloc->stream);
myfree((void *) alloc);
}
static void qmgr_transport_event(int unused_event, void *context)
{
QMGR_TRANSPORT_ALLOC *alloc = (QMGR_TRANSPORT_ALLOC *) context;
if (msg_verbose)
msg_info("transport_event: %s", alloc->transport->name);
event_cancel_timer(qmgr_transport_abort, context);
if (alloc->stream) {
event_disable_readwrite(vstream_fileno(alloc->stream));
non_blocking(vstream_fileno(alloc->stream), BLOCKING);
}
alloc->transport->pending -= 1;
if (alloc->transport->xport_rate_delay > 0) {
if ((alloc->transport->flags & QMGR_TRANSPORT_STAT_RATE_LOCK) == 0)
msg_panic("transport_event: missing rate lock for transport %s",
alloc->transport->name);
event_request_timer(qmgr_transport_rate_event, (void *) alloc,
alloc->transport->xport_rate_delay);
} else {
alloc->notify(alloc->transport, alloc->stream);
myfree((void *) alloc);
}
}
QMGR_TRANSPORT *qmgr_transport_select(void)
{
QMGR_TRANSPORT *xport;
QMGR_QUEUE *queue;
int need;
#define MIN5af51743e4eef(x, y) ((x) < (y) ? (x) : (y))
for (xport = qmgr_transport_list.next; xport; xport = xport->peers.next) {
if ((xport->flags & QMGR_TRANSPORT_STAT_DEAD) != 0
|| (xport->flags & QMGR_TRANSPORT_STAT_RATE_LOCK) != 0
|| xport->pending >= QMGR_TRANSPORT_MAX_PEND)
continue;
need = xport->pending + 1;
for (queue = xport->queue_list.next; queue; queue = queue->peers.next) {
if (QMGR_QUEUE_READY(queue) == 0)
continue;
if ((need -= MIN5af51743e4eef(queue->window - queue->busy_refcount,
queue->todo_refcount)) <= 0) {
QMGR_LIST_ROTATE(qmgr_transport_list, xport, peers);
if (msg_verbose)
msg_info("qmgr_transport_select: %s", xport->name);
return (xport);
}
}
}
return (0);
}
void qmgr_transport_alloc(QMGR_TRANSPORT *transport, QMGR_TRANSPORT_ALLOC_NOTIFY notify)
{
QMGR_TRANSPORT_ALLOC *alloc;
if (transport->flags & QMGR_TRANSPORT_STAT_DEAD)
msg_panic("qmgr_transport: dead transport: %s", transport->name);
if (transport->flags & QMGR_TRANSPORT_STAT_RATE_LOCK)
msg_panic("qmgr_transport: rate-locked transport: %s", transport->name);
if (transport->pending >= QMGR_TRANSPORT_MAX_PEND)
msg_panic("qmgr_transport: excess allocation: %s", transport->name);
if (transport->xport_rate_delay > 0)
transport->flags |= QMGR_TRANSPORT_STAT_RATE_LOCK;
alloc = (QMGR_TRANSPORT_ALLOC *) mymalloc(sizeof(*alloc));
alloc->transport = transport;
alloc->notify = notify;
transport->pending += 1;
if ((alloc->stream = mail_connect(MAIL_CLASS_PRIVATE, transport->name,
NON_BLOCKING)) == 0) {
msg_warn("connect to transport %s/%s: %m",
MAIL_CLASS_PRIVATE, transport->name);
event_request_timer(qmgr_transport_event, (void *) alloc, 0);
return;
}
#if (EVENTS_STYLE != EVENTS_STYLE_SELECT) && defined(CA_VSTREAM_CTL_DUPFD)
#ifndef THRESHOLD_FD_WORKAROUND
#define THRESHOLD_FD_WORKAROUND 128
#endif
vstream_control(alloc->stream,
CA_VSTREAM_CTL_DUPFD(THRESHOLD_FD_WORKAROUND),
CA_VSTREAM_CTL_END);
#endif
event_enable_read(vstream_fileno(alloc->stream), qmgr_transport_event,
(void *) alloc);
event_request_timer(qmgr_transport_abort, (void *) alloc,
var_daemon_timeout);
}
QMGR_TRANSPORT *qmgr_transport_create(const char *name)
{
QMGR_TRANSPORT *transport;
if (htable_find(qmgr_transport_byname, name) != 0)
msg_panic("qmgr_transport_create: transport exists: %s", name);
transport = (QMGR_TRANSPORT *) mymalloc(sizeof(QMGR_TRANSPORT));
transport->flags = 0;
transport->pending = 0;
transport->name = mystrdup(name);
transport->dest_concurrency_limit =
get_mail_conf_int2(name, _DEST_CON_LIMIT,
var_dest_con_limit, 0, 0);
transport->recipient_limit =
get_mail_conf_int2(name, _DEST_RCPT_LIMIT,
var_dest_rcpt_limit, 0, 0);
transport->init_dest_concurrency =
get_mail_conf_int2(name, _INIT_DEST_CON,
var_init_dest_concurrency, 1, 0);
transport->xport_rate_delay = get_mail_conf_time2(name, _XPORT_RATE_DELAY,
var_xport_rate_delay,
's', 0, 0);
transport->rate_delay = get_mail_conf_time2(name, _DEST_RATE_DELAY,
var_dest_rate_delay,
's', 0, 0);
if (transport->rate_delay > 0)
transport->dest_concurrency_limit = 1;
if (transport->dest_concurrency_limit != 0
&& transport->dest_concurrency_limit < transport->init_dest_concurrency)
transport->init_dest_concurrency = transport->dest_concurrency_limit;
transport->slot_cost = get_mail_conf_int2(name, _DELIVERY_SLOT_COST,
var_delivery_slot_cost, 0, 0);
transport->slot_loan = get_mail_conf_int2(name, _DELIVERY_SLOT_LOAN,
var_delivery_slot_loan, 0, 0);
transport->slot_loan_factor =
100 - get_mail_conf_int2(name, _DELIVERY_SLOT_DISCOUNT,
var_delivery_slot_discount, 0, 100);
transport->min_slots = get_mail_conf_int2(name, _MIN_DELIVERY_SLOTS,
var_min_delivery_slots, 0, 0);
transport->rcpt_unused = get_mail_conf_int2(name, _XPORT_RCPT_LIMIT,
var_xport_rcpt_limit, 0, 0);
transport->rcpt_per_stack = get_mail_conf_int2(name, _STACK_RCPT_LIMIT,
var_stack_rcpt_limit, 0, 0);
transport->refill_limit = get_mail_conf_int2(name, _XPORT_REFILL_LIMIT,
var_xport_refill_limit, 1, 0);
transport->refill_delay = get_mail_conf_time2(name, _XPORT_REFILL_DELAY,
var_xport_refill_delay, 's', 1, 0);
transport->queue_byname = htable_create(0);
QMGR_LIST_INIT(transport->queue_list);
transport->job_byname = htable_create(0);
QMGR_LIST_INIT(transport->job_list);
QMGR_LIST_INIT(transport->job_bytime);
transport->job_current = 0;
transport->job_next_unread = 0;
transport->candidate_cache = 0;
transport->candidate_cache_current = 0;
transport->candidate_cache_time = (time_t) 0;
transport->blocker_tag = 1;
transport->dsn = 0;
qmgr_feedback_init(&transport->pos_feedback, name, _CONC_POS_FDBACK,
VAR_CONC_POS_FDBACK, var_conc_pos_feedback);
qmgr_feedback_init(&transport->neg_feedback, name, _CONC_NEG_FDBACK,
VAR_CONC_NEG_FDBACK, var_conc_neg_feedback);
transport->fail_cohort_limit =
get_mail_conf_int2(name, _CONC_COHORT_LIM,
var_conc_cohort_limit, 0, 0);
if (qmgr_transport_byname == 0)
qmgr_transport_byname = htable_create(10);
htable_enter(qmgr_transport_byname, name, (void *) transport);
QMGR_LIST_PREPEND(qmgr_transport_list, transport, peers);
if (msg_verbose)
msg_info("qmgr_transport_create: %s concurrency %d recipients %d",
transport->name, transport->dest_concurrency_limit,
transport->recipient_limit);
return (transport);
}
QMGR_TRANSPORT *qmgr_transport_find(const char *name)
{
return ((QMGR_TRANSPORT *) htable_find(qmgr_transport_byname, name));
}