diff options
Diffstat (limited to '')
| -rw-r--r-- | src/qmgr/qmgr_job.c | 978 |
1 files changed, 978 insertions, 0 deletions
diff --git a/src/qmgr/qmgr_job.c b/src/qmgr/qmgr_job.c new file mode 100644 index 0000000..24065f3 --- /dev/null +++ b/src/qmgr/qmgr_job.c @@ -0,0 +1,978 @@ +/*++ +/* NAME +/* qmgr_job 3 +/* SUMMARY +/* per-transport jobs +/* SYNOPSIS +/* #include "qmgr.h" +/* +/* QMGR_JOB *qmgr_job_obtain(message, transport) +/* QMGR_MESSAGE *message; +/* QMGR_TRANSPORT *transport; +/* +/* void qmgr_job_free(job) +/* QMGR_JOB *job; +/* +/* void qmgr_job_move_limits(job) +/* QMGR_JOB *job; +/* +/* QMGR_ENTRY *qmgr_job_entry_select(transport) +/* QMGR_TRANSPORT *transport; +/* +/* void qmgr_job_blocker_update(queue) +/* QMGR_QUEUE *queue; +/* DESCRIPTION +/* These routines add/delete/manipulate per-transport jobs. +/* Each job corresponds to a specific transport and message. +/* Each job has a peer list containing all pending delivery +/* requests for that message. +/* +/* qmgr_job_obtain() finds an existing job for named message and +/* transport combination. New empty job is created if no existing can +/* be found. In either case, the job is prepared for assignment of +/* (more) message recipients. +/* +/* qmgr_job_free() disposes of a per-transport job after all +/* its entries have been taken care of. It is an error to dispose +/* of a job that is still in use. +/* +/* qmgr_job_entry_select() attempts to find the next entry suitable +/* for delivery. The job preempting algorithm is also exercised. +/* If necessary, an attempt to read more recipients into core is made. +/* This can result in creation of more job, queue and entry structures. +/* +/* qmgr_job_blocker_update() updates the status of blocked +/* jobs after a decrease in the queue's concurrency level, +/* after the queue is throttled, or after the queue is resumed +/* from suspension. +/* +/* qmgr_job_move_limits() takes care of proper distribution of the +/* per-transport recipients limit among the per-transport jobs. +/* Should be called whenever a job's recipient slot becomes available. +/* DIAGNOSTICS +/* Panic: consistency check failure. +/* LICENSE +/* .ad +/* .fi +/* The Secure Mailer license must be distributed with this software. +/* AUTHOR(S) +/* Patrik Rak +/* patrik@raxoft.cz +/*--*/ + +/* System library. */ + +#include <sys_defs.h> + +/* Utility library. */ + +#include <msg.h> +#include <htable.h> +#include <mymalloc.h> +#include <sane_time.h> + +/* Application-specific. */ + +#include "qmgr.h" + +/* Forward declarations */ + +static void qmgr_job_pop(QMGR_JOB *); + +/* Helper macros */ + +#define HAS_ENTRIES(job) ((job)->selected_entries < (job)->read_entries) + +/* + * The MIN_ENTRIES macro may underestimate a lot but we can't use message->rcpt_unread + * because we don't know if all those unread recipients go to our transport yet. + */ + +#define MIN_ENTRIES(job) ((job)->read_entries) +#define MAX_ENTRIES(job) ((job)->read_entries + (job)->message->rcpt_unread) + +#define RESET_CANDIDATE_CACHE(transport) ((transport)->candidate_cache_current = 0) + +#define IS_BLOCKER(job,transport) ((job)->blocker_tag == (transport)->blocker_tag) + +/* qmgr_job_create - create and initialize message job structure */ + +static QMGR_JOB *qmgr_job_create(QMGR_MESSAGE *message, QMGR_TRANSPORT *transport) +{ + QMGR_JOB *job; + + job = (QMGR_JOB *) mymalloc(sizeof(QMGR_JOB)); + job->message = message; + QMGR_LIST_APPEND(message->job_list, job, message_peers); + htable_enter(transport->job_byname, message->queue_id, (void *) job); + job->transport = transport; + QMGR_LIST_INIT(job->transport_peers); + QMGR_LIST_INIT(job->time_peers); + job->stack_parent = 0; + QMGR_LIST_INIT(job->stack_children); + QMGR_LIST_INIT(job->stack_siblings); + job->stack_level = -1; + job->blocker_tag = 0; + job->peer_byname = htable_create(0); + QMGR_LIST_INIT(job->peer_list); + job->slots_used = 0; + job->slots_available = 0; + job->selected_entries = 0; + job->read_entries = 0; + job->rcpt_count = 0; + job->rcpt_limit = 0; + return (job); +} + +/* qmgr_job_link - append the job to the job lists based on the time it was queued */ + +static void qmgr_job_link(QMGR_JOB *job) +{ + QMGR_TRANSPORT *transport = job->transport; + QMGR_MESSAGE *message = job->message; + QMGR_JOB *prev, *next, *list_prev, *list_next, *unread, *current; + int delay; + + /* + * Sanity checks. + */ + if (job->stack_level >= 0) + msg_panic("qmgr_job_link: already on the job lists (%d)", job->stack_level); + + /* + * Traverse the time list and the scheduler list from the end and stop + * when we found job older than the one being linked. + * + * During the traversals keep track if we have come across either the + * current job or the first unread job on the job list. If this is the + * case, these pointers will be adjusted below as required. + * + * Although both lists are exactly the same when only jobs on the stack + * level zero are considered, it's easier to traverse them separately. + * Otherwise it's impossible to keep track of the current job pointer + * effectively. + * + * This may look inefficient but under normal operation it is expected that + * the loops will stop right away, resulting in normal list appends + * below. However, this code is necessary for reviving retired jobs and + * for jobs which are created long after the first chunk of recipients + * was read in-core (either of these can happen only for multi-transport + * messages). + * + * XXX Note that we test stack_parent rather than stack_level below. This + * subtle difference allows us to enqueue the job in correct time order + * with respect to orphaned children even after their original parent on + * level zero is gone. Consequently, the early loop stop in candidate + * selection works reliably, too. These are the reasons why we care to + * bother with children adoption at all. + */ + current = transport->job_current; + for (next = 0, prev = transport->job_list.prev; prev; + next = prev, prev = prev->transport_peers.prev) { + if (prev->stack_parent == 0) { + delay = message->queued_time - prev->message->queued_time; + if (delay >= 0) + break; + } + if (current == prev) + current = 0; + } + list_prev = prev; + list_next = next; + + unread = transport->job_next_unread; + for (next = 0, prev = transport->job_bytime.prev; prev; + next = prev, prev = prev->time_peers.prev) { + delay = message->queued_time - prev->message->queued_time; + if (delay >= 0) + break; + if (unread == prev) + unread = 0; + } + + /* + * Link the job into the proper place on the job lists and mark it so we + * know it has been linked. + */ + job->stack_level = 0; + QMGR_LIST_LINK(transport->job_list, list_prev, job, list_next, transport_peers); + QMGR_LIST_LINK(transport->job_bytime, prev, job, next, time_peers); + + /* + * Update the current job pointer if necessary. + */ + if (current == 0) + transport->job_current = job; + + /* + * Update the pointer to the first unread job on the job list and steal + * the unused recipient slots from the old one. + */ + if (unread == 0) { + unread = transport->job_next_unread; + transport->job_next_unread = job; + if (unread != 0) + qmgr_job_move_limits(unread); + } + + /* + * Get as much recipient slots as possible. The excess will be returned + * to the transport pool as soon as the exact amount required is known + * (which is usually after all recipients have been read in core). + */ + if (transport->rcpt_unused > 0) { + job->rcpt_limit += transport->rcpt_unused; + message->rcpt_limit += transport->rcpt_unused; + transport->rcpt_unused = 0; + } +} + +/* qmgr_job_find - lookup job associated with named message and transport */ + +static QMGR_JOB *qmgr_job_find(QMGR_MESSAGE *message, QMGR_TRANSPORT *transport) +{ + + /* + * Instead of traversing the message job list, we use single per + * transport hash table. This is better (at least with respect to memory + * usage) than having single hash table (usually almost empty) for each + * message. + */ + return ((QMGR_JOB *) htable_find(transport->job_byname, message->queue_id)); +} + +/* qmgr_job_obtain - find/create the appropriate job and make it ready for new recipients */ + +QMGR_JOB *qmgr_job_obtain(QMGR_MESSAGE *message, QMGR_TRANSPORT *transport) +{ + QMGR_JOB *job; + + /* + * Try finding an existing job, reviving it if it was already retired. + * Create a new job for this transport/message combination otherwise. In + * either case, the job ends linked on the job lists. + */ + if ((job = qmgr_job_find(message, transport)) == 0) + job = qmgr_job_create(message, transport); + if (job->stack_level < 0) + qmgr_job_link(job); + + /* + * Reset the candidate cache because of the new expected recipients. Make + * sure the job is not marked as a blocker for the same reason. Note that + * this can result in having a non-blocker followed by more blockers. + * Consequently, we can't just update the current job pointer, we have to + * reset it. Fortunately qmgr_job_entry_select() will easily deal with + * this and will lookup the real current job for us. + */ + RESET_CANDIDATE_CACHE(transport); + if (IS_BLOCKER(job, transport)) { + job->blocker_tag = 0; + transport->job_current = transport->job_list.next; + } + return (job); +} + +/* qmgr_job_move_limits - move unused recipient slots to the next unread job */ + +void qmgr_job_move_limits(QMGR_JOB *job) +{ + QMGR_TRANSPORT *transport = job->transport; + QMGR_MESSAGE *message = job->message; + QMGR_JOB *next = transport->job_next_unread; + int rcpt_unused, msg_rcpt_unused; + + /* + * Find next unread job on the job list if necessary. Cache it for later. + * This makes the amortized efficiency of this routine O(1) per job. Note + * that we use the time list whose ordering doesn't change over time. + */ + if (job == next) { + for (next = next->time_peers.next; next; next = next->time_peers.next) + if (next->message->rcpt_offset != 0) + break; + transport->job_next_unread = next; + } + + /* + * Calculate the number of available unused slots. + */ + rcpt_unused = job->rcpt_limit - job->rcpt_count; + msg_rcpt_unused = message->rcpt_limit - message->rcpt_count; + if (msg_rcpt_unused < rcpt_unused) + rcpt_unused = msg_rcpt_unused; + + /* + * Transfer the unused recipient slots back to the transport pool and to + * the next not-fully-read job. Job's message limits are adjusted + * accordingly. Note that the transport pool can be negative if we used + * some of the rcpt_per_stack slots. + */ + if (rcpt_unused > 0) { + job->rcpt_limit -= rcpt_unused; + message->rcpt_limit -= rcpt_unused; + transport->rcpt_unused += rcpt_unused; + if (next != 0 && (rcpt_unused = transport->rcpt_unused) > 0) { + next->rcpt_limit += rcpt_unused; + next->message->rcpt_limit += rcpt_unused; + transport->rcpt_unused = 0; + } + } +} + +/* qmgr_job_parent_gone - take care of orphaned stack children */ + +static void qmgr_job_parent_gone(QMGR_JOB *job, QMGR_JOB *parent) +{ + QMGR_JOB *child; + + while ((child = job->stack_children.next) != 0) { + QMGR_LIST_UNLINK(job->stack_children, QMGR_JOB *, child, stack_siblings); + if (parent != 0) + QMGR_LIST_APPEND(parent->stack_children, child, stack_siblings); + child->stack_parent = parent; + } +} + +/* qmgr_job_unlink - unlink the job from the job lists */ + +static void qmgr_job_unlink(QMGR_JOB *job) +{ + const char *myname = "qmgr_job_unlink"; + QMGR_TRANSPORT *transport = job->transport; + + /* + * Sanity checks. + */ + if (job->stack_level != 0) + msg_panic("%s: non-zero stack level (%d)", myname, job->stack_level); + if (job->stack_parent != 0) + msg_panic("%s: parent present", myname); + if (job->stack_siblings.next != 0) + msg_panic("%s: siblings present", myname); + + /* + * Make sure that children of job on zero stack level are informed that + * their parent is gone too. + */ + qmgr_job_parent_gone(job, 0); + + /* + * Update the current job pointer if necessary. + */ + if (transport->job_current == job) + transport->job_current = job->transport_peers.next; + + /* + * Invalidate the candidate selection cache if necessary. + */ + if (job == transport->candidate_cache + || job == transport->candidate_cache_current) + RESET_CANDIDATE_CACHE(transport); + + /* + * Remove the job from the job lists and mark it as unlinked. + */ + QMGR_LIST_UNLINK(transport->job_list, QMGR_JOB *, job, transport_peers); + QMGR_LIST_UNLINK(transport->job_bytime, QMGR_JOB *, job, time_peers); + job->stack_level = -1; +} + +/* qmgr_job_retire - remove the job from the job lists while waiting for recipients to deliver */ + +static void qmgr_job_retire(QMGR_JOB *job) +{ + if (msg_verbose) + msg_info("qmgr_job_retire: %s", job->message->queue_id); + + /* + * Pop the job from the job stack if necessary. + */ + if (job->stack_level > 0) + qmgr_job_pop(job); + + /* + * Make sure this job is not cached as the next unread job for this + * transport. The qmgr_entry_done() will make sure that the slots donated + * by this job are moved back to the transport pool as soon as possible. + */ + qmgr_job_move_limits(job); + + /* + * Remove the job from the job lists. Note that it remains on the message + * job list, though, and that it can be revived by using + * qmgr_job_obtain(). Also note that the available slot counter is left + * intact. + */ + qmgr_job_unlink(job); +} + +/* qmgr_job_free - release the job structure */ + +void qmgr_job_free(QMGR_JOB *job) +{ + const char *myname = "qmgr_job_free"; + QMGR_MESSAGE *message = job->message; + QMGR_TRANSPORT *transport = job->transport; + + if (msg_verbose) + msg_info("%s: %s %s", myname, message->queue_id, transport->name); + + /* + * Sanity checks. + */ + if (job->rcpt_count) + msg_panic("%s: non-zero recipient count (%d)", myname, job->rcpt_count); + + /* + * Pop the job from the job stack if necessary. + */ + if (job->stack_level > 0) + qmgr_job_pop(job); + + /* + * Return any remaining recipient slots back to the recipient slots pool. + */ + qmgr_job_move_limits(job); + if (job->rcpt_limit) + msg_panic("%s: recipient slots leak (%d)", myname, job->rcpt_limit); + + /* + * Unlink and discard the structure. Check if the job is still linked on + * the job lists or if it was already retired before unlinking it. + */ + if (job->stack_level >= 0) + qmgr_job_unlink(job); + QMGR_LIST_UNLINK(message->job_list, QMGR_JOB *, job, message_peers); + htable_delete(transport->job_byname, message->queue_id, (void (*) (void *)) 0); + htable_free(job->peer_byname, (void (*) (void *)) 0); + myfree((void *) job); +} + +/* qmgr_job_count_slots - maintain the delivery slot counters */ + +static void qmgr_job_count_slots(QMGR_JOB *job) +{ + + /* + * Count the number of delivery slots used during the delivery of the + * selected job. Also count the number of delivery slots available for + * its preemption. + * + * Despite its trivial look, this is one of the key parts of the theory + * behind this preempting scheduler. + */ + job->slots_available++; + job->slots_used++; + + /* + * If the selected job is not the original current job, reset the + * candidate cache because the change above have slightly increased the + * chance of this job becoming a candidate next time. + * + * Don't expect that the change of the current jobs this turn will render + * the candidate cache invalid the next turn - it can happen that the + * next turn the original current job will be selected again and the + * cache would be considered valid in such case. + */ + if (job != job->transport->candidate_cache_current) + RESET_CANDIDATE_CACHE(job->transport); +} + +/* qmgr_job_candidate - find best job candidate for preempting given job */ + +static QMGR_JOB *qmgr_job_candidate(QMGR_JOB *current) +{ + QMGR_TRANSPORT *transport = current->transport; + QMGR_JOB *job, *best_job = 0; + double score, best_score = 0.0; + int max_slots, max_needed_entries, max_total_entries; + int delay; + time_t now = sane_time(); + + /* + * Fetch the result directly from the cache if the cache is still valid. + * + * Note that we cache negative results too, so the cache must be invalidated + * by resetting the cached current job pointer, not the candidate pointer + * itself. + * + * In case the cache is valid and contains no candidate, we can ignore the + * time change, as it affects only which candidate is the best, not if + * one exists. However, this feature requires that we no longer relax the + * cache resetting rules, depending on the automatic cache timeout. + */ + if (transport->candidate_cache_current == current + && (transport->candidate_cache_time == now + || transport->candidate_cache == 0)) + return (transport->candidate_cache); + + /* + * Estimate the minimum amount of delivery slots that can ever be + * accumulated for the given job. All jobs that won't fit into these + * slots are excluded from the candidate selection. + */ + max_slots = (MIN_ENTRIES(current) - current->selected_entries + + current->slots_available) / transport->slot_cost; + + /* + * Select the candidate with best time_since_queued/total_recipients + * score. In addition to jobs which don't meet the max_slots limit, skip + * also jobs which don't have any selectable entries at the moment. + * + * Instead of traversing the whole job list we traverse it just from the + * current job forward. This has several advantages. First, we skip some + * of the blocker jobs and the current job itself right away. But the + * really important advantage is that we are sure that we don't consider + * any jobs that are already stack children of the current job. Thanks to + * this we can easily include all encountered jobs which are leaf + * children of some of the preempting stacks as valid candidates. All we + * need to do is to make sure we do not include any of the stack parents. + * And, because the leaf children are not ordered by the time since + * queued, we have to exclude them from the early loop end test. + * + * However, don't bother searching if we can't find anything suitable + * anyway. + */ + if (max_slots > 0) { + for (job = current->transport_peers.next; job; job = job->transport_peers.next) { + if (job->stack_children.next != 0 || IS_BLOCKER(job, transport)) + continue; + max_total_entries = MAX_ENTRIES(job); + max_needed_entries = max_total_entries - job->selected_entries; + delay = now - job->message->queued_time + 1; + if (max_needed_entries > 0 && max_needed_entries <= max_slots) { + score = (double) delay / max_total_entries; + if (score > best_score) { + best_score = score; + best_job = job; + } + } + + /* + * Stop early if the best score is as good as it can get. + */ + if (delay <= best_score && job->stack_level == 0) + break; + } + } + + /* + * Cache the result for later use. + */ + transport->candidate_cache = best_job; + transport->candidate_cache_current = current; + transport->candidate_cache_time = now; + + return (best_job); +} + +/* qmgr_job_preempt - preempt large message with smaller one */ + +static QMGR_JOB *qmgr_job_preempt(QMGR_JOB *current) +{ + const char *myname = "qmgr_job_preempt"; + QMGR_TRANSPORT *transport = current->transport; + QMGR_JOB *job, *prev; + int expected_slots; + int rcpt_slots; + + /* + * Suppress preempting completely if the current job is not big enough to + * accumulate even the minimal number of slots required. + * + * Also, don't look for better job candidate if there are no available slots + * yet (the count can get negative due to the slot loans below). + */ + if (current->slots_available <= 0 + || MAX_ENTRIES(current) < transport->min_slots * transport->slot_cost) + return (current); + + /* + * Find best candidate for preempting the current job. + * + * Note that the function also takes care that the candidate fits within the + * number of delivery slots which the current job is still able to + * accumulate. + */ + if ((job = qmgr_job_candidate(current)) == 0) + return (current); + + /* + * Sanity checks. + */ + if (job == current) + msg_panic("%s: attempt to preempt itself", myname); + if (job->stack_children.next != 0) + msg_panic("%s: already on the job stack (%d)", myname, job->stack_level); + if (job->stack_level < 0) + msg_panic("%s: not on the job list (%d)", myname, job->stack_level); + + /* + * Check if there is enough available delivery slots accumulated to + * preempt the current job. + * + * The slot loaning scheme improves the average message response time. Note + * that the loan only allows the preemption happen earlier, though. It + * doesn't affect how many slots have to be "paid" - in either case the + * full number of slots required has to be accumulated later before the + * current job can be preempted again. + */ + expected_slots = MAX_ENTRIES(job) - job->selected_entries; + if (current->slots_available / transport->slot_cost + transport->slot_loan + < expected_slots * transport->slot_loan_factor / 100.0) + return (current); + + /* + * Preempt the current job. + * + * This involves placing the selected candidate in front of the current job + * on the job list and updating the stack parent/child/sibling pointers + * appropriately. But first we need to make sure that the candidate is + * taken from its previous job stack which it might be top of. + */ + if (job->stack_level > 0) + qmgr_job_pop(job); + QMGR_LIST_UNLINK(transport->job_list, QMGR_JOB *, job, transport_peers); + prev = current->transport_peers.prev; + QMGR_LIST_LINK(transport->job_list, prev, job, current, transport_peers); + job->stack_parent = current; + QMGR_LIST_APPEND(current->stack_children, job, stack_siblings); + job->stack_level = current->stack_level + 1; + + /* + * Update the current job pointer and explicitly reset the candidate + * cache. + */ + transport->job_current = job; + RESET_CANDIDATE_CACHE(transport); + + /* + * Since the single job can be preempted by several jobs at the same + * time, we have to adjust the available slot count now to prevent using + * the same slots multiple times. To do that we subtract the number of + * slots the preempting job will supposedly use. This number will be + * corrected later when that job is popped from the stack to reflect the + * number of slots really used. + * + * As long as we don't need to keep track of how many slots were really + * used, we can (ab)use the slots_used counter for counting the + * difference between the real and expected amounts instead of the + * absolute amount. + */ + current->slots_available -= expected_slots * transport->slot_cost; + job->slots_used = -expected_slots; + + /* + * Add part of extra recipient slots reserved for preempting jobs to the + * new current job if necessary. + * + * Note that transport->rcpt_unused is within <-rcpt_per_stack,0> in such + * case. + */ + if (job->message->rcpt_offset != 0) { + rcpt_slots = (transport->rcpt_per_stack + transport->rcpt_unused + 1) / 2; + job->rcpt_limit += rcpt_slots; + job->message->rcpt_limit += rcpt_slots; + transport->rcpt_unused -= rcpt_slots; + } + if (msg_verbose) + msg_info("%s: %s by %s, level %d", myname, current->message->queue_id, + job->message->queue_id, job->stack_level); + + return (job); +} + +/* qmgr_job_pop - remove the job from its job preemption stack */ + +static void qmgr_job_pop(QMGR_JOB *job) +{ + const char *myname = "qmgr_job_pop"; + QMGR_TRANSPORT *transport = job->transport; + QMGR_JOB *parent; + + if (msg_verbose) + msg_info("%s: %s", myname, job->message->queue_id); + + /* + * Sanity checks. + */ + if (job->stack_level <= 0) + msg_panic("%s: not on the job stack (%d)", myname, job->stack_level); + + /* + * Adjust the number of delivery slots available to preempt job's parent. + * Note that the -= actually adds back any unused slots, as we have + * already subtracted the expected amount of slots from both counters + * when we did the preemption. + * + * Note that we intentionally do not adjust slots_used of the parent. Doing + * so would decrease the maximum per message inflation factor if the + * preemption appeared near the end of parent delivery. + * + * For the same reason we do not adjust parent's slots_available if the + * parent is not the original parent that was preempted by this job + * (i.e., the original parent job has already completed). + * + * This is another key part of the theory behind this preempting scheduler. + */ + if ((parent = job->stack_parent) != 0 + && job->stack_level == parent->stack_level + 1) + parent->slots_available -= job->slots_used * transport->slot_cost; + + /* + * Remove the job from its parent's children list. + */ + if (parent != 0) { + QMGR_LIST_UNLINK(parent->stack_children, QMGR_JOB *, job, stack_siblings); + job->stack_parent = 0; + } + + /* + * If there is a parent, let it adopt all those orphaned children. + * Otherwise at least notify the children that their parent is gone. + */ + qmgr_job_parent_gone(job, parent); + + /* + * Put the job back to stack level zero. + */ + job->stack_level = 0; + + /* + * Explicitly reset the candidate cache. It's not worth trying to skip + * this under some complicated conditions - in most cases the popped job + * is the current job so we would have to reset it anyway. + */ + RESET_CANDIDATE_CACHE(transport); + + /* + * Here we leave the remaining work involving the proper placement on the + * job list to the caller. The most important reason for this is that it + * allows us not to look up where exactly to place the job. + * + * The caller is also made responsible for invalidating the current job + * cache if necessary. + */ +#if 0 + QMGR_LIST_UNLINK(transport->job_list, QMGR_JOB *, job, transport_peers); + QMGR_LIST_LINK(transport->job_list, some_prev, job, some_next, transport_peers); + + if (transport->job_current == job) + transport->job_current = job->transport_peers.next; +#endif +} + +/* qmgr_job_peer_select - select next peer suitable for delivery */ + +static QMGR_PEER *qmgr_job_peer_select(QMGR_JOB *job) +{ + QMGR_PEER *peer; + QMGR_MESSAGE *message = job->message; + + /* + * Try reading in more recipients. We do that as soon as possible + * (almost, see below), to make sure there is enough new blood pouring + * in. Otherwise single recipient for slow destination might starve the + * entire message delivery, leaving lot of fast destination recipients + * sitting idle in the queue file. + * + * Ideally we would like to read in recipients whenever there is a space, + * but to prevent excessive I/O, we read them only when enough time has + * passed or we can read enough of them at once. + * + * Note that even if we read the recipients few at a time, the message + * loading code tries to put them to existing recipient entries whenever + * possible, so the per-destination recipient grouping is not grossly + * affected. + * + * XXX Workaround for logic mismatch. The message->refcount test needs + * explanation. If the refcount is zero, it means that qmgr_active_done() + * is being completed asynchronously. In such case, we can't read in + * more recipients as bad things would happen after qmgr_active_done() + * continues processing. Note that this results in the given job being + * stalled for some time, but fortunately this particular situation is so + * rare that it is not critical. Still we seek for better solution. + */ + if (message->rcpt_offset != 0 + && message->refcount > 0 + && (message->rcpt_limit - message->rcpt_count >= job->transport->refill_limit + || (message->rcpt_limit > message->rcpt_count + && sane_time() - message->refill_time >= job->transport->refill_delay))) + qmgr_message_realloc(message); + + /* + * Get the next suitable peer, if there is any. + */ + if (HAS_ENTRIES(job) && (peer = qmgr_peer_select(job)) != 0) + return (peer); + + /* + * There is no suitable peer in-core, so try reading in more recipients + * if possible. This is our last chance to get suitable peer before + * giving up on this job for now. + * + * XXX For message->refcount, see above. + */ + if (message->rcpt_offset != 0 + && message->refcount > 0 + && message->rcpt_limit > message->rcpt_count) { + qmgr_message_realloc(message); + if (HAS_ENTRIES(job)) + return (qmgr_peer_select(job)); + } + return (0); +} + +/* qmgr_job_entry_select - select next entry suitable for delivery */ + +QMGR_ENTRY *qmgr_job_entry_select(QMGR_TRANSPORT *transport) +{ + QMGR_JOB *job, *next; + QMGR_PEER *peer; + QMGR_ENTRY *entry; + + /* + * Get the current job if there is one. + */ + if ((job = transport->job_current) == 0) + return (0); + + /* + * Exercise the preempting algorithm if enabled. + * + * The slot_cost equal to 1 causes the algorithm to degenerate and is + * therefore disabled too. + */ + if (transport->slot_cost >= 2) + job = qmgr_job_preempt(job); + + /* + * Select next entry suitable for delivery. In case the current job can't + * provide one because of the per-destination concurrency limits, we mark + * it as a "blocker" job and continue with the next job on the job list. + * + * Note that the loop also takes care of getting the "stall" jobs (job with + * no entries currently available) out of the way if necessary. Stall + * jobs can appear in case of multi-transport messages whose recipients + * don't fit in-core at once. Some jobs created by such message may have + * only few recipients and would stay on the job list until all other + * jobs of that message are delivered, blocking precious recipient slots + * available to this transport. Or it can happen that the job has some + * more entries but suddenly they all get deferred. Whatever the reason, + * we retire such jobs below if we happen to come across some. + */ + for ( /* empty */ ; job; job = next) { + next = job->transport_peers.next; + + /* + * Don't bother if the job is known to have no available entries + * because of the per-destination concurrency limits. + */ + if (IS_BLOCKER(job, transport)) + continue; + + if ((peer = qmgr_job_peer_select(job)) != 0) { + + /* + * We have found a suitable peer. Select one of its entries and + * adjust the delivery slot counters. + */ + entry = qmgr_entry_select(peer); + qmgr_job_count_slots(job); + + /* + * Remember the current job for the next time so we don't have to + * crawl over all those blockers again. They will be reconsidered + * when the concurrency limit permits. + */ + transport->job_current = job; + + /* + * In case we selected the very last job entry, remove the job + * from the job lists right now. + * + * This action uses the assumption that once the job entry has been + * selected, it can be unselected only before the message ifself + * is deferred. Thus the job with all entries selected can't + * re-appear with more entries available for selection again + * (without reading in more entries from the queue file, which in + * turn invokes qmgr_job_obtain() which re-links the job back on + * the lists if necessary). + * + * Note that qmgr_job_move_limits() transfers the recipients slots + * correctly even if the job is unlinked from the job list thanks + * to the job_next_unread caching. + */ + if (!HAS_ENTRIES(job) && job->message->rcpt_offset == 0) + qmgr_job_retire(job); + + /* + * Finally. Hand back the fruit of our tedious effort. + */ + return (entry); + } else if (HAS_ENTRIES(job)) { + + /* + * The job can't be selected due the concurrency limits. Mark it + * together with its queues so we know they are blocking the job + * list and they get the appropriate treatment. In particular, + * all blockers will be reconsidered when one of the problematic + * queues will accept more deliveries. And the job itself will be + * reconsidered if it is assigned some more entries. + */ + job->blocker_tag = transport->blocker_tag; + for (peer = job->peer_list.next; peer; peer = peer->peers.next) + if (peer->entry_list.next != 0) + peer->queue->blocker_tag = transport->blocker_tag; + } else { + + /* + * The job is "stalled". Retire it until it either gets freed or + * gets more entries later. + */ + qmgr_job_retire(job); + } + } + + /* + * We have not found any entry we could use for delivery. Well, things + * must have changed since this transport was selected for asynchronous + * allocation. Never mind. Clear the current job pointer and reluctantly + * report back that we have failed in our task. + */ + transport->job_current = 0; + return (0); +} + +/* qmgr_job_blocker_update - update "blocked job" status */ + +void qmgr_job_blocker_update(QMGR_QUEUE *queue) +{ + QMGR_TRANSPORT *transport = queue->transport; + + /* + * If the queue was blocking some of the jobs on the job list, check if + * the concurrency limit has lifted. If there are still some pending + * deliveries, give it a try and unmark all transport blockers at once. + * The qmgr_job_entry_select() will do the rest. In either case make sure + * the queue is not marked as a blocker anymore, with extra handling of + * queues which were declared dead. + * + * Note that changing the blocker status also affects the candidate cache. + * Most of the cases would be automatically recognized by the current job + * change, but we play safe and reset the cache explicitly below. + * + * Keeping the transport blocker tag odd is an easy way to make sure the tag + * never matches jobs that are not explicitly marked as blockers. + */ + if (queue->blocker_tag == transport->blocker_tag) { + if (queue->window > queue->busy_refcount && queue->todo.next != 0) { + transport->blocker_tag += 2; + transport->job_current = transport->job_list.next; + transport->candidate_cache_current = 0; + } + if (queue->window > queue->busy_refcount || QMGR_QUEUE_THROTTLED(queue)) + queue->blocker_tag = 0; + } +} |