Limit the number of simultaneous connect attempts passed to libpurple.

[thrasher.git] / perl / lib / Thrasher / ConnectionManager.pm

package Thrasher::ConnectionManager;
use strict;
use warnings;

=pod

=head1 NAME

Thrasher::ConnectionManager - manage the act of connecting to the
legacy services

=head1 DESCRIPTION

Thrasher::ConnectionManager centralizes and separates the management
of whether or not we want to connect to a legacy server. Likely this
will get more complicated later.

Suppose we have a couple hundred users on the gateway when the 
connection is lost to the legacy service, that is, I<entirely> lost.
Later, it comes back. We don't want to hammer the service with 
repeated connection attempts from each user. 

We also want to be able to control the rate at which re-connection
attempts are made. If someone tries to log on during the period when
the connection is down, we don't really want them to add to the
failure proceedings. 

::ConnectionManager implements the ability to schedule connection
attempts, collect information about whether the logins are successful
or not, and decide when to attempt to log in again.

This module sort of functions as a singleton (sort of), exporting the
following function:

=over 4

=item *

C<request_connect>($connection_closure): Request that the given
closure be run at some point in the future. This closure should 
I<eventually> result in a call to C<connection_success> or
C<connection_failure>, which may itself arise from a callback
or something (like for a connection), which is why we can't 
just use the return value. The connection manager has no
way to enforce this.

A connection is successful if the user successfully logged on.
That is, a password failure is a failure too. This avoids the
case where we could hammer a service that was having 
authentication problems.

=back

=cut

use Carp qw(confess);

our $NOW = undef;

# Setting this to 1 will simply bypass this entire thing.
our $IGNORE_CONNECTION_MANAGER = 0;

use base 'Exporter';
our @EXPORT_OK = qw(decayable connection_success connection_failure
                    request_connect schedule_request);
our %EXPORT_TAGS = (all => \@EXPORT_OK);

use Thrasher::Log qw(log debug);
require Thrasher::Component;

# $SIMULTANEOUS_CONNECT_BOUND's initial value sets the maximum number
# of simultaneous, possibly asynchronous, connection attempts. It will
# be decremented each time a connection attempt starts and incremented
# when it finishes. Thus, at any given time while the
# ConnectionManager is managing it will be the current number of new
# connect attempts that could start at that time.
#
# This should usually be a low limit. Most protocols have several DNS
# lookups and TCP connections per IM login.
our $SIMULTANEOUS_CONNECT_BOUND = 2;

sub decayable {
    return new Thrasher::ConnectionManager::Decayable(@_);
}

# So, the design goals:
# * If there's no reason to suspect network troubles, connect away.
# * If there's a total network disconnect, a bare minimum of
#   connection attempts should be made, not a straight
#   (number of logins) * (constant frequency) * time
# * It can't take too long to recover from a total network failure.
#   Even for hundreds of people, it should recover relatively quickly.
# * If at all possible, this should adapt to rate limiting. (It may
#   not be possible if the legacy service simply cuts of your
#   service, rather than metering it.)
#
# Approach: We define a "decayable number", which exponentially 
# decays over time, as seen in
# Thrasher::ConnectionManager::Decayable. As successful connections
# are made, the count of successful connections is incremented. As
# unsuccessful connections are made, the count of unsuccessful 
# connections is incremented. The ratio of the two is used to 
# determine connection time, bounded on both ends by a limit
# (if it's below a certain number, we simply connect immediately,
# if it's above a certain number we assume it needs to be bounded
# to something like one attempt every 30 seconds). We also, after
# a certain point, start queuing up connection requests. This way, the
# system reacts to the current apparent state of the connection
# to the legacy service without hammering on the legacy system, 
# which may be considered hostile.


### TUNABLE PARAMETERS

# This is the base decay rate applied to our two counters,
# applied directly to the failure and indirectly to the success
# counter.
our $decay_rate = .65;

# Optimism is how much more likely we are to think that we can
# connect again. Values about that cause the system to "remember"
# success more strongly than failure.
our $optimism = 3;

my $log30 = CORE::log(30);

### Values

our $success = decayable(5, 1 - ((1-$decay_rate) / $optimism));
our $failure = decayable(0, $decay_rate);

# This is incremented once per connection attempt and is used to
# limit the total rate of reconnect attempts.
# Please, Thrasher, don't hammer 'em.
our $hammering = decayable(0, .05);
our $hammering_limit = 15;

# Things queued up because we can't try to connect yet.
my @queue;

my $next_connection_attempt_time;

# This takes the clousure to run and the time to run it. This
# allows you to either call out to the event loop, or, in testing,
# override this and capture the execution attempts to verify
# the time is correct.
our $scheduler = sub {
    my $closure = shift;
    my $timeout = shift;

    return $Thrasher::Component::COMPONENT->{'event_loop'}->schedule($closure,
                                                                     $timeout);
};

# "If I make a request right now, what will the delay and
# forcibly_enqueue values be?"
sub schedule_request {
    # Check that we aren't hammering the remote server.
    if ($hammering->value > $hammering_limit) {
        log("Hammering: " . $hammering->value . " of $hammering_limit."
            ." Delaying connection request.");
        return [0, 1];
    }

    my $current_success = $success->value;
    my $current_failure = $failure->value;

    if ($current_failure < 0.0001) {
        log("Never failed, immediately connecting.");
        return [0, 0];
    } 
    if ($current_success < 0.00001 && $current_failure > 0.00001) {
        log("Can't remember having had failure or success, immediately connecting.");
        return [0, 1];
    }
    if ($current_success < 0.0001) {
        log("Can't remember having succeeded ($current_success), "
            ."but I remember failure ($current_failure). Max delay.");
        return [30, 0];
    }
    
    my $ratio = $current_failure / $current_success;

    if ($ratio > $log30) {
        log("Failure/success ratio: $ratio, ($current_failure / $current_success), capped to 30 second delay");
        return [30, 0];
    }
    
    my $delay = exp($ratio) - 1;
    log("Failure/success ratio: $ratio ($current_failure / $current_success), comes out to $delay for delay");
    return [$delay, 0];
}

# This returns true if the connection will be immediate, 
# otherwise it returns false.
sub request_connect {
    my $connection_closure = shift;

    # Deal with this when we once again reschedule things.
    if (defined($next_connection_attempt_time) &&
        ($NOW || time) < $next_connection_attempt_time) {
        push @queue, $connection_closure;
        return 0;
    }

    if ($SIMULTANEOUS_CONNECT_BOUND <= 0) {
        log('Enough connect attempts in progress; delaying.');
        push(@queue, $connection_closure);
        schedule_connection_executor(1);
        return 0;
    }
    elsif (!@queue) {
        my ($delay, $forcibly_enqueue) = @{schedule_request()};

        # Catch this in the next scheduling run.
        if ($forcibly_enqueue) {
            push @queue, $connection_closure;
            schedule_connection_executor(1);
            return 0;
        }
        
        # Little to no delay called for, just go for it
        # (leaning on hammering protection to prevent hammering)
        if ($delay < 1) {
            service_request($connection_closure);
            return 1;
        }

        # We're not in a hammering delay and we need to
        # reschedule to delay beyond a second
        push @queue, $connection_closure;
        schedule_connection_executor($delay);
        return 0;
    } else {
        push @queue, $connection_closure;
        return 0;
    }
}

# schedule_connection_executor($timeout_in_seconds)
my $run_scheduled = 0;
sub schedule_connection_executor {
    my $timeout = shift;

    if ($run_scheduled) {
        # request_connect() called in between scheduled
        # execution_runner() calls.
        return;
    }

    $next_connection_attempt_time = ($NOW || time) + $timeout;

    my $execution_runner = sub {
        log(scalar(@queue) . ' connection attempts queued.');
        if (@queue && $SIMULTANEOUS_CONNECT_BOUND <= 0) {
            debug('Enough connect attempts in progress (not rescheduled).');
            return 1;
        }

        my $new_timeout = eval {
            return connection_executor();
        };
        if ($@) {
            $run_scheduled = 0;
            die;
        }
        elsif (! $new_timeout) {
            debug("Executor done.");
            $run_scheduled = 0;
            return 0;
        }
        elsif ($new_timeout == $timeout) {
            debug("Next executor in $timeout seconds (not rescheduled).");
            return 1;
        }
        else {
            debug("Next executor in $new_timeout seconds (rescheduled).");
            $run_scheduled = 0;
            schedule_connection_executor($new_timeout);
            return 0;
        }
    };

    $run_scheduled = $scheduler->($execution_runner,
                                  $timeout * 1000);
}

# connection_executor() is what is run by the scheduler, which will
# propogate itself until it runs out of work.
#
# Returns the new $timeout_in_seconds until the desired next run, or
# false when no next run should be scheduled.
sub connection_executor {
    # If we're still in a hammering lock, delay another second
    if ($hammering->value >= $hammering_limit) {
        return 1;
    }

    if (!@queue) {
        # We've finally emptied the queue. Resume normal usage.
        return 0;
    }

    elsif ($SIMULTANEOUS_CONNECT_BOUND <= 0) {
        # Delay repeat until some connection attempts finish.
        return 1;
    }

    # Otherwise, it's time to take one thing off the queue and run it.
    my $current_closure = shift @queue;
    service_request($current_closure);

    # And now it's time to schedule the next request
    my ($delay, $forcibly_enqueue) = @{schedule_request()};
    if ($forcibly_enqueue) {
        return 1;
    }

    if ($delay < 1) {
        goto &connection_executor;
    }

    # Otherwise, reschedule $delay seconds
    return $delay;
}

sub service_request {
    my $closure = shift;
    $hammering->add(1);
    local $@;

    if (!defined($closure)) {
        confess "Undefined closure passed to service_request.";
    }
    log("Servicing connection request.");
    eval {
        $closure->();
    };
    if ($@) {
        log("Failure in connection callback: $@");
    }
    else {
        $SIMULTANEOUS_CONNECT_BOUND--;
    }
}

sub connection_success {
    $SIMULTANEOUS_CONNECT_BOUND++;

    debug('Connection success managed.');
    $success->add(1);
}

sub connection_failure {
    $SIMULTANEOUS_CONNECT_BOUND++;

    debug('Connection failure managed.');
    my $current_schedule = schedule_request;
    if ($current_schedule->[0] != 30) {
        $failure->add(10);
    } else {
        log("Declining to increment failure.");
    }
}

package Thrasher::ConnectionManager::Decayable;
use strict;
use warnings;

# This implements a "decayable" number. It starts with the value
# you give it, and is multiplied by the decay rate raise to the
# power of the number of minutes since the last time it was queried. 
# The fact that this tends to decay to zero is desired.
# Example: new Thrasher::ConnectionManager::Decayable(16, .5)
# will be 8 in one minute, 4 in the next, 2 in the next, and so on.
# The transition is smooth; it will be 11.3137... after 30 seconds.

use Time::HiRes qw(time tv_interval);

sub new {
    my $class = shift;
    my $current_value = shift;
    my $decay_rate = shift;
    my $now = $Thrasher::ConnectionManager::NOW || scalar(time);

    my $decayable = { value => $current_value,
                      decay_rate => $decay_rate,
                      last_computed => $now };
    bless $decayable, $class;
    
    return $decayable;
}

sub value {
    my $self = shift;
    my $now = $Thrasher::ConnectionManager::NOW || scalar(time);
    
    my $since_last_computation = $now - $self->{last_computed};
    my $minutes = $since_last_computation / 60;

    my $decay = $self->{decay_rate} ** $minutes;
    my $current_value = $self->{value} * $decay;

    $self->{value} = $current_value;
    $self->{last_computed} = $now;

    return $self->{value};
}

sub add {
    my $self = shift;
    my $inc = shift;

    my $value = $self->value;
    $self->{value} = $value + $inc;

    return $self->{value};
}

sub subtract {
    my $self = shift;
    my $dec = shift;
    $self->add(-$dec);
}

1;