trAvis - MANAGER

Edit File: AsyncLoop.pm

# <@LICENSE>
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to you under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at:
# 
#     http://www.apache.org/licenses/LICENSE-2.0
# 
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# </@LICENSE>

=head1 NAME

Mail::SpamAssassin::AsyncLoop - scanner asynchronous event loop

=head1 DESCRIPTION

An asynchronous event loop used for long-running operations, performed "in the
background" during the Mail::SpamAssassin::check() scan operation, such as DNS
blocklist lookups.

=head1 METHODS

=over 4

=cut

package Mail::SpamAssassin::AsyncLoop;

use strict;
use warnings;
use bytes;
use re 'taint';

use Time::HiRes qw(time);

use Mail::SpamAssassin;
use Mail::SpamAssassin::Logger;

our @ISA = qw();

# obtain timer resolution if possible
BEGIN {
  use vars qw($timer_resolution);
  eval {
    $timer_resolution = Time::HiRes->can('clock_getres')
      ? Time::HiRes::clock_getres(Time::HiRes::CLOCK_REALTIME())
      : 0.001;  # wild guess, assume resolution is better than 1s
    1;
  } or do {
    $timer_resolution = 1;  # Perl's builtin timer ticks at one second
  };
}

#############################################################################

sub new {
  # called from PerMsgStatus, a new AsyncLoop object is created
  # for each new message processing
  my $class = shift;
  $class = ref($class) || $class;

my ($main) = @_;
  my $self = {
    main                => $main,
    queries_started     => 0,
    queries_completed   => 0,
    total_queries_started   => 0,
    total_queries_completed => 0,
    pending_lookups     => { },
    timing_by_query     => { },
    all_lookups         => { },  # keyed by "rr_type/domain"
  };

bless ($self, $class);
  $self;
}

# Given a domain name, produces a listref of successively stripped down
# parent domains, e.g. a domain '2.10.Example.COM' would produce a list:
# '2.10.example.com', '10.example.com', 'example.com', 'com', ''
#
sub domain_to_search_list {
  my ($domain) = @_;
  $domain =~ s/^\.+//; $domain =~ s/\.+\z//;  # strip leading and trailing dots
  my @search_keys;
  if ($domain =~ /\[/) {  # don't split address literals
    @search_keys = ( $domain, '' );  # presumably an address literal
  } else {
    local $1;
    $domain = lc $domain;
    for (;;) {
      push(@search_keys, $domain);
      last  if $domain eq '';
      # strip one level
      $domain = ($domain =~ /^ (?: [^.]* ) \. (.*) \z/xs) ? $1 : '';
    }
    if (@search_keys > 20) {  # enforce some sanity limit
      @search_keys = @search_keys[$#search_keys-19 .. $#search_keys];
    }
  }
  return \@search_keys;
}

# ---------------------------------------------------------------------------

=item $ent = $async->start_lookup($ent, $master_deadline)

Register the start of a long-running asynchronous lookup operation.
C<$ent> is a hash reference containing the following items:

=over 4

=item key (required)

A key string, unique to this lookup.  This is what is reported in
debug messages, used as the key for C<get_lookup()>, etc.

=item id (required)

An ID string, also unique to this lookup.  Typically, this is the DNS packet ID
as returned by DnsResolver's C<bgsend> method.  Sadly, the Net::DNS
architecture forces us to keep a separate ID string for this task instead of
reusing C<key> -- if you are not using DNS lookups through DnsResolver, it
should be OK to just reuse C<key>.

=item type (required)

A string, typically one word, used to describe the type of lookup in log
messages, such as C<DNSBL>, C<MX>, C<TXT>.

=item zone (optional)

A zone specification (typically a DNS zone name - e.g. host, domain, or RBL)
which may be used as a key to look up per-zone settings. No semantics on this
parameter is imposed by this module. Currently used to fetch by-zone timeouts.

=item timeout_initial (optional)

An initial value of elapsed time for which we are willing to wait for a
response (time in seconds, floating point value is allowed). When elapsed
time since a query started exceeds the timeout value and there are no other
queries to wait for, the query is aborted. The actual timeout value ranges
from timeout_initial and gradually approaches timeout_min (see next parameter)
as the number of already completed queries approaches the number of all
queries started.

If a caller does not explicitly provide this parameter or its value is
undefined, a default initial timeout value is settable by a configuration
variable rbl_timeout.

If a value of the timeout_initial parameter is below timeout_min, the initial
timeout is set to timeout_min.

=item timeout_min (optional)

A lower bound (in seconds) to which the actual timeout approaches as the
number of queries completed approaches the number of all queries started.
Defaults to 0.2 * timeout_initial.

=back

C<$ent> is returned by this method, with its contents augmented by additional
information.

=cut

sub start_lookup {
  my ($self, $ent, $master_deadline) = @_;

my $id  = $ent->{id};
  my $key = $ent->{key};
  defined $id && $id ne ''  or die "oops, no id";
  $key                      or die "oops, no key";
  $ent->{type}              or die "oops, no type";

my $now = time;
  $ent->{start_time} = $now  if !defined $ent->{start_time};

# are there any applicable per-zone settings?
  my $zone = $ent->{zone};
  my $settings;  # a ref to a by-zone or to global settings
  my $conf_by_zone = $self->{main}->{conf}->{by_zone};
  if (defined $zone && $conf_by_zone) {
  # dbg("async: searching for by_zone settings for $zone");
    $zone =~ s/^\.//;  $zone =~ s/\.\z//;  # strip leading and trailing dot
    for (;;) {  # 2.10.example.com, 10.example.com, example.com, com, ''
      if (exists $conf_by_zone->{$zone}) {
        $settings = $conf_by_zone->{$zone};
        last;
      } elsif ($zone eq '') {
        last;
      } else {  # strip one level, careful with address literals
        $zone = ($zone =~ /^( (?: [^.] | \[ (?: \\. | [^\]\\] )* \] )* )
                            \. (.*) \z/xs) ? $2 : '';
      }
    }
  }

dbg("async: applying by_zone settings for %s", $zone)  if $settings;

my $t_init = $ent->{timeout_initial};  # application-specified has precedence
  $t_init = $settings->{rbl_timeout}  if $settings && !defined $t_init;
  $t_init = $self->{main}->{conf}->{rbl_timeout}  if !defined $t_init;
  $t_init = 0  if !defined $t_init;      # last-resort default, just in case

my $t_end = $ent->{timeout_min};       # application-specified has precedence
  $t_end = $settings->{rbl_timeout_min}  if $settings && !defined $t_end;
  $t_end = 0.2 * $t_init  if !defined $t_end;
  $t_end = 0  if $t_end < 0;  # just in case
  $t_init = $t_end  if $t_init < $t_end;

my $clipped_by_master_deadline = 0;
  if (defined $master_deadline) {
    my $time_avail = $master_deadline - time;
    $time_avail = 0.5  if $time_avail < 0.5;  # give some slack
    if ($t_init > $time_avail) {
      $t_init = $time_avail; $clipped_by_master_deadline = 1;
      $t_end  = $time_avail  if $t_end > $time_avail;
    }
  }
  $ent->{timeout_initial} = $t_init;
  $ent->{timeout_min} = $t_end;

$ent->{display_id} =  # identifies entry in debug logging and similar
    join(", ", grep { defined }
               map { ref $ent->{$_} ? @{$ent->{$_}} : $ent->{$_} }
               qw(sets rules rulename type key) );

$self->{pending_lookups}->{$key} = $ent;

$self->{queries_started}++;
  $self->{total_queries_started}++;
  dbg("async: starting: %s (timeout %.1fs, min %.1fs)%s",
      $ent->{display_id}, $ent->{timeout_initial}, $ent->{timeout_min},
      !$clipped_by_master_deadline ? '' : ', capped by time limit');

$ent;
}

# ---------------------------------------------------------------------------

=item $ent = $async->bgsend_and_start_lookup($domain, $type, $class, $ent, $cb, %options)

A common idiom: calls C<bgsend>, followed by a call to C<start_lookup>,
returning the argument $ent object as modified by C<start_lookup> and
filled-in with a query ID.

=cut

sub bgsend_and_start_lookup {
  my($self, $domain, $type, $class, $ent, $cb, %options) = @_;
  $ent = {}  if !$ent;
  $domain =~ s/\.+\z//s;  # strip trailing dots, these sometimes still sneak in
  $ent->{id} = undef;
  $ent->{query_type} = $type;
  $ent->{query_domain} = $domain;
  $ent->{type} = $type  if !exists $ent->{type};
  $cb = $ent->{completed_callback}  if !$cb;  # compatibility with SA < 3.4

my $key = $ent->{key} || '';

my $dnskey = uc($type) . '/' . lc($domain);
  my $dns_query_info = $self->{all_lookups}{$dnskey};

if ($dns_query_info) {  # DNS query already underway or completed
    my $id = $ent->{id} = $dns_query_info->{id};  # re-use existing query
    return if !defined $id;  # presumably blocked, or other fatal failure
    my $id_tail = $id; $id_tail =~ s{^\d+/IN/}{};
    lc($id_tail) eq lc($dnskey)
      or info("async: unmatched id %s, key=%s", $id, $dnskey);

my $pkt = $dns_query_info->{pkt};
    if (!$pkt) {  # DNS query underway, still waiting for results
      # just add our query to the existing one
      push(@{$dns_query_info->{applicants}}, [$ent,$cb]);
      dbg("async: query %s already underway, adding no.%d %s",
          $id, scalar @{$dns_query_info->{applicants}},
          $ent->{rulename} || $key);

} else {  # DNS query already completed, re-use results
      # answer already known, just do the callback and be done with it
      if (!$cb) {
        dbg("async: query %s already done, re-using for %s", $id, $key);
      } else {
        dbg("async: query %s already done, re-using for %s, callback",
            $id, $key);
        eval {
          $cb->($ent, $pkt); 1;
        } or do {
          chomp $@;
          # resignal if alarm went off
          die "async: (1) $@\n"  if $@ =~ /__alarm__ignore__$.*$/s;
          warn sprintf("query %s completed, callback %s failed: %s\n",
                       $id, $key, $@);
        };
      }
    }
  }

else {  # no existing query, open a new DNS query
    $dns_query_info = $self->{all_lookups}{$dnskey} = {};  # new query needed
    my($id, $blocked);
    my $dns_query_blockages = $self->{main}->{conf}->{dns_query_blocked};
    if ($dns_query_blockages) {
      my $search_list = domain_to_search_list($domain);
      foreach my $parent_domain (@$search_list) {
        $blocked = $dns_query_blockages->{$parent_domain};
        last if defined $blocked; # stop at first defined, can be true or false
      }
    }
    if ($blocked) {
      dbg("async: blocked by dns_query_restriction: %s", $dnskey);
    } else {
      dbg("async: launching %s for %s", $dnskey, $key);
      $id = $self->{main}->{resolver}->bgsend($domain, $type, $class, sub {
          my($pkt, $pkt_id, $timestamp) = @_;
          # this callback sub is called from DnsResolver::poll_responses()
        # dbg("async: in a bgsend_and_start_lookup callback, id %s", $pkt_id);
          if ($pkt_id ne $id) {
            warn "async: mismatched dns id: got $pkt_id, expected $id\n";
            return;
          }
          $self->set_response_packet($pkt_id, $pkt, $ent->{key}, $timestamp);
          $dns_query_info->{pkt} = $pkt;
          my $cb_count = 0;
          foreach my $tuple (@{$dns_query_info->{applicants}}) {
            my($appl_ent, $appl_cb) = @$tuple;
            if ($appl_cb) {
              dbg("async: calling callback on key %s%s", $key,
                  !defined $appl_ent->{rulename} ? ''
                    : ", rule ".$appl_ent->{rulename});
              $cb_count++;
              eval {
                $appl_cb->($appl_ent, $pkt); 1;
              } or do {
                chomp $@;
                # resignal if alarm went off
                die "async: (2) $@\n"  if $@ =~ /__alarm__ignore__$.*$/s;
                warn sprintf("query %s completed, callback %s failed: %s\n",
                             $id, $appl_ent->{key}, $@);
              };
            }
          }
          delete $dns_query_info->{applicants};
          dbg("async: query $id completed, no callbacks run")  if !$cb_count;
        });
    }
    return if !defined $id;
    $dns_query_info->{id} = $ent->{id} = $id;
    push(@{$dns_query_info->{applicants}}, [$ent,$cb]);
    $self->start_lookup($ent, $options{master_deadline});
  }
  return $ent;
}

# ---------------------------------------------------------------------------

=item $ent = $async->get_lookup($key)

Retrieve the pending-lookup object for the given key C<$key>.

If the lookup is complete, this will return C<undef>.

Note that a lookup is still considered "pending" until C<complete_lookups()> is
called, even if it has been reported as complete via C<set_response_packet()>.

=cut

sub get_lookup {
  my ($self, $key) = @_;
  return $self->{pending_lookups}->{$key};
}

# ---------------------------------------------------------------------------

=item $async->log_lookups_timing()

Log sorted timing for all completed lookups.

=cut

sub log_lookups_timing {
  my ($self) = @_;
  my $timings = $self->{timing_by_query};
  for my $key (sort { $timings->{$a} <=> $timings->{$b} } keys %$timings) {
    dbg("async: timing: %.3f %s", $timings->{$key}, $key);
  }
}

# ---------------------------------------------------------------------------

=item $alldone = $async->complete_lookups()

Perform a poll of the pending lookups, to see if any are completed.
Callbacks on completed queries will be called from poll_responses().

If there are no lookups remaining, or if too much time has elapsed since
any results were returned, C<1> is returned, otherwise C<0>.

=cut

sub complete_lookups {
  my ($self, $timeout, $allow_aborting_of_expired) = @_;
  my $alldone = 0;
  my $anydone = 0;
  my $allexpired = 1;
  my %typecount;

my $pending = $self->{pending_lookups};
  $self->{queries_started} = 0;
  $self->{queries_completed} = 0;

my $now = time;

if (defined $timeout && $timeout > 0 &&
      %$pending && $self->{total_queries_started} > 0)
  {
    # shrink a 'select' timeout if a caller specified unnecessarily long
    # value beyond the latest deadline of any outstanding request;
    # can save needless wait time (up to 1 second in harvest_dnsbl_queries)
    my $r = $self->{total_queries_completed} / $self->{total_queries_started};
    my $r2 = $r * $r;  # 0..1
    my $max_deadline;
    while (my($key,$ent) = each %$pending) {
      my $t_init = $ent->{timeout_initial};
      my $dt = $t_init - ($t_init - $ent->{timeout_min}) * $r2;
      my $deadline = $ent->{start_time} + $dt;
      $max_deadline = $deadline  if !defined $max_deadline ||
                                    $deadline > $max_deadline;
    }
    if (defined $max_deadline) {
      # adjust to timer resolution, only deals with 1s and with fine resolution
      $max_deadline = 1 + int $max_deadline
        if $timer_resolution == 1 && $max_deadline > int $max_deadline;
      my $sufficient_timeout = $max_deadline - $now;
      $sufficient_timeout = 0  if $sufficient_timeout < 0;
      if ($timeout > $sufficient_timeout) {
        dbg("async: reducing select timeout from %.1f to %.1f s",
            $timeout, $sufficient_timeout);
        $timeout = $sufficient_timeout;
      }
    }
  }

# trap this loop in an eval { } block, as Net::DNS could throw
  # die()s our way; in particular, process_dnsbl_results() has
  # thrown die()s before (bug 3794).
  eval {

if (%$pending) {  # any outstanding requests still?
      $self->{last_poll_responses_time} = $now;
      my $nfound = $self->{main}->{resolver}->poll_responses($timeout);
      dbg("async: select found %s responses ready (t.o.=%.1f)",
          !$nfound ? 'no' : $nfound,  $timeout);
    }
    $now = time;  # capture new timestamp, after possible sleep in 'select'

# A callback routine may generate another DNS query, which may insert
    # an entry into the %$pending hash thus invalidating the each() context.
    # So, make sure that callbacks are not called while the each() context
    # is open. [Bug 6937]
    #
    while (my($key,$ent) = each %$pending) {
      my $id = $ent->{id};
      if (exists $self->{finished}->{$id}) {
        delete $self->{finished}->{$id};
        $anydone = 1;
        $ent->{finish_time} = $now  if !defined $ent->{finish_time};
        my $elapsed = $ent->{finish_time} - $ent->{start_time};
        dbg("async: completed in %.3f s: %s", $elapsed, $ent->{display_id});
        $self->{timing_by_query}->{". $key"} += $elapsed;
        $self->{queries_completed}++;
        $self->{total_queries_completed}++;
        delete $pending->{$key};
      }
    }

if (%$pending) {  # still any requests outstanding? are they expired?
      my $r =
        !$allow_aborting_of_expired || !$self->{total_queries_started} ? 1.0
        : $self->{total_queries_completed} / $self->{total_queries_started};
      my $r2 = $r * $r;  # 0..1
      while (my($key,$ent) = each %$pending) {
        $typecount{$ent->{type}}++;
        my $t_init = $ent->{timeout_initial};
        my $dt = $t_init - ($t_init - $ent->{timeout_min}) * $r2;
        # adjust to timer resolution, only deals with 1s and fine resolution
        $dt = 1 + int $dt  if $timer_resolution == 1 && $dt > int $dt;
        $allexpired = 0  if $now <= $ent->{start_time} + $dt;
      }
      dbg("async: queries completed: %d, started: %d",
          $self->{queries_completed}, $self->{queries_started});
    }

# ensure we don't get stuck if a request gets lost in the ether.
    if (! %$pending) {
      $alldone = 1;
    }
    elsif ($allexpired && $allow_aborting_of_expired) {
      # avoid looping forever if we haven't got all results.
      dbg("async: escaping: lost or timed out requests or responses");
      $self->abort_remaining_lookups();
      $alldone = 1;
    }
    else {
      dbg("async: queries active: %s%s at %s",
          join (' ', map { "$_=$typecount{$_}" } sort keys %typecount),
          $allexpired ? ', all expired' : '', scalar(localtime(time)));
      $alldone = 0;
    }
    1;

} or do {
    my $eval_stat = $@ ne '' ? $@ : "errno=$!";  chomp $eval_stat;
    # resignal if alarm went off
    die "async: (3) $eval_stat\n"  if $eval_stat =~ /__alarm__ignore__$.*$/s;
    dbg("async: caught complete_lookups death, aborting: %s", $eval_stat);
    $alldone = 1;      # abort remaining
  };

return wantarray ? ($alldone,$anydone) : $alldone;
}

# ---------------------------------------------------------------------------

=item $async->abort_remaining_lookups()

Abort any remaining lookups.

=cut

sub abort_remaining_lookups {
  my ($self) = @_;

my $pending = $self->{pending_lookups};
  my $foundcnt = 0;
  my $now = time;

while (my($key,$ent) = each %$pending) {
    dbg("async: aborting after %.3f s, %s: %s",
        $now - $ent->{start_time},
        (defined $ent->{timeout_initial} &&
         $now > $ent->{start_time} + $ent->{timeout_initial}
           ? 'past original deadline' : 'deadline shrunk'),
        $ent->{display_id} );
    $foundcnt++;
    $self->{timing_by_query}->{"X $key"} = $now - $ent->{start_time};
    $ent->{finish_time} = $now  if !defined $ent->{finish_time};
    delete $pending->{$key};
  }

# call any remaining callbacks, indicating the query has been aborted
  #
  my $all_lookups_ref = $self->{all_lookups};
  foreach my $dnskey (keys %$all_lookups_ref) {
    my $dns_query_info = $all_lookups_ref->{$dnskey};
    my $cb_count = 0;
    foreach my $tuple (@{$dns_query_info->{applicants}}) {
      my($ent, $cb) = @$tuple;
      if ($cb) {
        dbg("async: calling callback/abort on key %s%s", $dnskey,
            !defined $ent->{rulename} ? '' : ", rule ".$ent->{rulename});
        $cb_count++;
        eval {
          $cb->($ent, undef); 1;
        } or do {
          chomp $@;
          # resignal if alarm went off
          die "async: (2) $@\n"  if $@ =~ /__alarm__ignore__$.*$/s;
          warn sprintf("query %s aborted, callback %s failed: %s\n",
                       $dnskey, $ent->{key}, $@);
        };
      }
      dbg("async: query $dnskey aborted, no callbacks run")  if !$cb_count;
    }
    delete $dns_query_info->{applicants};
  }

dbg("async: aborted %d remaining lookups", $foundcnt)  if $foundcnt > 0;
  delete $self->{last_poll_responses_time};
  $self->{main}->{resolver}->bgabort();
  1;
}

# ---------------------------------------------------------------------------

=item $async->set_response_packet($id, $pkt, $key, $timestamp)

Register a "response packet" for a given query.  C<$id> is the ID for the
query, and must match the C<id> supplied in C<start_lookup()>. C<$pkt> is the
packet object for the response. A parameter C<$key> identifies an entry in a
hash %{$self->{pending_lookups}} where the object which spawned this query can
be found, and through which futher information about the query is accessible.

C<$pkt> may be undef, indicating that no response packet is available, but a
query has completed (e.g. was aborted or dismissed) and is no longer "pending".

The DNS resolver's response packet C<$pkt> will be made available to a callback
subroutine through its argument as well as in C<$ent-<gt>{response_packet}>.

=cut

sub set_response_packet {
  my ($self, $id, $pkt, $key, $timestamp) = @_;
  $self->{finished}->{$id} = 1;  # only key existence matters, any value
  $timestamp = time  if !defined $timestamp;
  my $pending = $self->{pending_lookups};
  if (!defined $key) {  # backward compatibility with 3.2.3 and older plugins
    # a third-party plugin did not provide $key in a call, search for it:
    if ($id eq $pending->{$id}->{id}) {  # I feel lucky, key==id ?
      $key = $id;
    } else {  # then again, maybe not, be more systematic
      for my $tkey (keys %$pending) {
        if ($id eq $pending->{$tkey}->{id}) { $key = $tkey; last }
      }
    }
    dbg("async: got response on id $id, search found key $key");
  }
  if (!defined $key) {
    info("async: no key, response packet not remembered, id $id");
  } else {
    my $ent = $pending->{$key};
    my $ent_id = $ent->{id};
    if (!defined $ent_id) {
      # should not happen, troubleshooting
      info("async: ignoring response, id %s, ent_id is undef: %s",
           $id, join(', ', %$ent));
    } elsif ($id ne $ent_id) {
      info("async: ignoring response, mismatched id $id, expected $ent_id");
    } else {
      $ent->{finish_time} = $timestamp;
      $ent->{response_packet} = $pkt;
    }
  }
  1;
}

=item $async->report_id_complete($id,$key,$key,$timestamp)

Legacy. Equivalent to $self->set_response_packet($id,undef,$key,$timestamp),
i.e. providing undef as a response packet. Register that a query has
completed and is no longer "pending". C<$id> is the ID for the query,
and must match the C<id> supplied in C<start_lookup()>.

One or the other of C<set_response_packet()> or C<report_id_complete()>
should be called, but not both.

=cut

sub report_id_complete {
  my ($self, $id, $key, $timestamp) = @_;
  $self->set_response_packet($id, undef, $key, $timestamp);
}

# ---------------------------------------------------------------------------

=item $time = $async->last_poll_responses_time()

Get the time of the last call to C<poll_responses()> (which is called
from C<complete_lookups()>.  If C<poll_responses()> was never called or
C<abort_remaining_lookups()> has been called C<last_poll_responses_time()>
will return undef.

=cut

sub last_poll_responses_time {
  my ($self) = @_;
  return $self->{last_poll_responses_time};
}

=back

=cut