File Coverage

blib/lib/YAHC.pm

Criterion	Covered	Total	%
statement	380	629	60.4
branch	189	474	39.8
condition	72	202	35.6
subroutine	52	84	61.9
pod	26	26	100.0
total	719	1415	50.8

package YAHC;

use strict;
use warnings;

our $VERSION = '0.033';

use EV;
use Time::HiRes;
use Exporter 'import';
use Scalar::Util qw/weaken/;
use Fcntl qw/F_GETFL F_SETFL O_NONBLOCK/;
use POSIX qw/EINPROGRESS EINTR EAGAIN EWOULDBLOCK strftime/;
use Socket qw/PF_INET SOCK_STREAM $CRLF SOL_SOCKET SO_ERROR inet_aton inet_ntoa pack_sockaddr_in/;
use constant SSL => $ENV{YAHC_NO_SSL} ? 0 : eval 'use IO::Socket::SSL 1.94 (); 1';
use constant SSL_WANT_READ  => SSL ? IO::Socket::SSL::SSL_WANT_READ()  : 0;
use constant SSL_WANT_WRITE => SSL ? IO::Socket::SSL::SSL_WANT_WRITE() : 0;

sub YAHC::Error::NO_ERROR                () { 0 }
sub YAHC::Error::REQUEST_TIMEOUT         () { 1 << 0 }
sub YAHC::Error::CONNECT_TIMEOUT         () { 1 << 1 }
sub YAHC::Error::DRAIN_TIMEOUT           () { 1 << 2 }
sub YAHC::Error::LIFETIME_TIMEOUT        () { 1 << 3 }
sub YAHC::Error::TIMEOUT                 () { 1 << 8 }
sub YAHC::Error::RETRY_LIMIT             () { 1 << 9 }

sub YAHC::Error::CONNECT_ERROR           () { 1 << 10 }
sub YAHC::Error::READ_ERROR              () { 1 << 11 }
sub YAHC::Error::WRITE_ERROR             () { 1 << 12 }
sub YAHC::Error::REQUEST_ERROR           () { 1 << 13 }
sub YAHC::Error::RESPONSE_ERROR          () { 1 << 14 }
sub YAHC::Error::CALLBACK_ERROR          () { 1 << 15 }
sub YAHC::Error::SSL_ERROR               () { 1 << 16 }
sub YAHC::Error::TERMINAL_ERROR          () { 1 << 30 }
sub YAHC::Error::INTERNAL_ERROR          () { 1 << 31 }

sub YAHC::State::INITIALIZED             () { 0   }
sub YAHC::State::RESOLVE_DNS             () { 5   }
sub YAHC::State::CONNECTING              () { 10  }
sub YAHC::State::CONNECTED               () { 15  }
sub YAHC::State::SSL_HANDSHAKE           () { 20  }
sub YAHC::State::WRITING                 () { 25  }
sub YAHC::State::READING                 () { 30  }
sub YAHC::State::USER_ACTION             () { 35  }
sub YAHC::State::COMPLETED               () { 100 } # terminal state

sub YAHC::SocketCache::GET               () { 1 }
sub YAHC::SocketCache::STORE             () { 2 }

use constant {
    # TCP_READ_CHUNK should *NOT* be lower than 16KB because of SSL things.
    # https://metacpan.org/pod/distribution/IO-Socket-SSL/lib/IO/Socket/SSL.pod
    # Another way might be if you try to sysread at least 16kByte all the time.
    # 16kByte is the maximum size of an SSL frame and because sysread returns
    # data from only a single SSL frame you can guarantee that there are no
    # pending data.
    TCP_READ_CHUNK              => 131072,
    CALLBACKS                   => [ qw/init_callback connecting_callback connected_callback
                                        writing_callback reading_callback callback/ ],
};

our @EXPORT_OK = qw/
    yahc_retry_conn
    yahc_reinit_conn
    yahc_terminal_error
    yahc_conn_last_error
    yahc_conn_id
    yahc_conn_url
    yahc_conn_target
    yahc_conn_state
    yahc_conn_errors
    yahc_conn_timeline
    yahc_conn_request
    yahc_conn_response
    yahc_conn_attempt
    yahc_conn_attempts_left
    yahc_conn_socket_cache_id
    yahc_conn_register_error
    yahc_conn_user_data
/;

our %EXPORT_TAGS = (all => \@EXPORT_OK);
my $LAST_CONNECTION_ID;

################################################################################
# User facing functons
################################################################################

sub new {
    my ($class, $args) = @_;
    $LAST_CONNECTION_ID = $$ * 1000 unless defined $LAST_CONNECTION_ID;

    die 'YAHC: ->new() expect args to be a hashref' if defined $args and ref($args) ne 'HASH';
    die 'YAHC: please do `my ($yahc, $yahc_storage) = YAHC::new()` and keep both these objects in the same scope' unless wantarray;

    # wrapping target selection here allows all client share same list
    # and more importantly to share index within the list
    $args->{_target}  = _wrap_host(delete $args->{host})             if $args->{host};
    $args->{_backoff} = _wrap_backoff(delete $args->{backoff_delay}) if $args->{backoff_delay};
    $args->{_socket_cache} = _wrap_socket_cache(delete $args->{socket_cache}) if $args->{socket_cache};

    my %storage;
    my $self = bless {
        loop                => new EV::Loop,
        pid                 => $$, # store pid to detect forks
        storage             => \%storage,
        debug               => delete $args->{debug} || $ENV{YAHC_DEBUG} || 0,
        keep_timeline       => delete $args->{keep_timeline} || $ENV{YAHC_TIMELINE} || 0,
        pool_args           => $args,
    }, $class;

    # this's a radical way of avoiding circular references.
    # let's see how it plays out in practise.
    weaken($self->{storage});
    weaken($self->{$_} = $storage{$_} = {}) for qw/watchers callbacks connections/;

    if (delete $args->{account_for_signals}) {
        _log_message('YAHC: enable account_for_signals logic') if $self->{debug};
        my $sigcheck = $self->{watchers}{_sigcheck} = $self->{loop}->check(sub {});
        $sigcheck->keepalive(0);
    }

    return $self, \%storage;
}

sub request {
    my ($self, @args) = @_;
    die 'YAHC: new_request() expects arguments' unless @args;
    die 'YAHC: storage object is destroyed' unless $self->{storage};

    my ($conn_id, $request) = (@args == 1 ? ('connection_' . $LAST_CONNECTION_ID++, $args[0]) : @args);
    die "YAHC: Connection with name '$conn_id' already exists\n"
        if exists $self->{connections}{$conn_id};

    my $pool_args = $self->{pool_args};
    do { $request->{$_} ||= $pool_args->{$_} if $pool_args->{$_} } foreach (qw/host port scheme head
                                                                               request_timeout connect_timeout
                                                                               drain_timeout lifetime_timeout/);
    if ($request->{host}) {
        $request->{_target} = _wrap_host($request->{host});
    } elsif ($pool_args->{_target}) {
        $request->{_target} = $pool_args->{_target};
    } else {
        die "YAHC: host must be defined in request() or in new()\n";
    }

    if ($request->{backoff_delay}) {
        $request->{_backoff} = _wrap_backoff($request->{backoff_delay});
    } elsif ($pool_args->{_backoff}) {
        $request->{_backoff} = $pool_args->{_backoff};
    }

    if ($request->{socket_cache}) {
        $request->{_socket_cache} = _wrap_socket_cache($request->{socket_cache});
    } elsif ($pool_args->{_socket_cache}) {
        $request->{_socket_cache} = $pool_args->{_socket_cache};
    }

    my $scheme = $request->{scheme} ||= 'http';
    my $debug = delete $request->{debug} || $self->{debug};
    my $keep_timeline = delete $request->{keep_timeline} || $self->{keep_timeline};
    my $user_data = delete $request->{user_data};

    my $conn = {
        id          => $conn_id,
        request     => $request,
        response    => { status => 0 },
        attempt     => 0,
        retries     => $request->{retries} || 0,
        state       => YAHC::State::INITIALIZED(),
        selected_target => [],
        ($debug                   ? (debug => $debug) : ()),
        ($keep_timeline           ? (keep_timeline => $keep_timeline) : ()),
        ($debug || $keep_timeline ? (debug_or_timeline => 1) : ()),
        (defined $user_data       ? (user_data => $user_data) : ()),
        pid         => $$,
    };

    my %callbacks;
    foreach (@{ CALLBACKS() }) {
        next unless exists $request->{$_};
        my $cb = $callbacks{$_} = delete $request->{$_};
        $conn->{"has_$_"} = !!$cb;
    }

    $self->{watchers}{$conn_id} = {};
    $self->{callbacks}{$conn_id} = \%callbacks;
    $self->{connections}{$conn_id} = $conn;

    _set_lifetime_timer($self, $conn_id) if $request->{lifetime_timeout};

    return $conn if $request->{_test}; # for testing purposes
    _set_init_state($self, $conn_id);

    # if user fire new request in a callback we need to update stop_condition
    my $stop_condition = $self->{stop_condition};
    if ($stop_condition && $stop_condition->{all}) {
        $stop_condition->{connections}{$conn_id} = 1;
    }

    return $conn;
}

sub drop {
    my ($self, $c, $force_socket_close) = @_;
    my $conn_id = ref($c) eq 'HASH' ? $c->{id} : $c;
    my $conn = $self->{connections}{$conn_id} or return;
    _register_in_timeline($conn, "dropping connection from pool") if exists $conn->{debug_or_timeline};
    _set_completed_state($self, $conn_id, $force_socket_close) unless $conn->{state} == YAHC::State::COMPLETED();
    return $conn;
}

sub run         { shift->_run(0, @_)            }
sub run_once    { shift->_run(EV::RUN_ONCE)     }
sub run_tick    { shift->_run(EV::RUN_NOWAIT)   }
sub is_running  { !!shift->{loop}->depth        }
sub loop        { shift->{loop}                 }

sub break {
    my ($self, $reason) = @_;
    return unless $self->is_running;
    _log_message('YAHC: pid %d breaking event loop because %s', $$, ($reason || 'no reason')) if $self->{debug};
    $self->{loop}->break(EV::BREAK_ONE)
}

################################################################################
# Routines to manipulate connections (also user facing)
################################################################################

sub yahc_terminal_error {
    return (($_[0] & YAHC::Error::TERMINAL_ERROR()) == YAHC::Error::TERMINAL_ERROR()) ? 1 : 0;
}

sub yahc_reinit_conn {
    my ($conn, $args) = @_;
    die "YAHC: cannot reinit completed connection\n"
        if $conn->{state} >= YAHC::State::COMPLETED();

    $conn->{attempt} = 0;
    $conn->{state} = YAHC::State::INITIALIZED();
    return unless defined $args && ref($args) eq 'HASH';

    my $request = $conn->{request};
    $request->{_target}  = _wrap_host(delete $args->{host})             if $args->{host};
    $request->{_backoff} = _wrap_backoff(delete $args->{backoff_delay}) if $args->{backoff_delay};
    do { $request->{$_} = $args->{$_} if $args->{$_} } foreach (keys %$args);
}

sub yahc_retry_conn {
    my ($conn, $args) = @_;
    die "YAHC: cannot retry completed connection\n"
        if $conn->{state} >= YAHC::State::COMPLETED();
    return unless yahc_conn_attempts_left($conn) > 0;

    $conn->{state} = YAHC::State::INITIALIZED();
    return unless defined $args && ref($args) eq 'HASH';

    $conn->{request}{_backoff} = _wrap_backoff($args->{backoff_delay})
        if $args->{backoff_delay};
}

sub yahc_conn_last_error {
    my $conn = shift;
    return unless $conn->{errors} && @{ $conn->{errors} };
    return wantarray ? @{ $conn->{errors}[-1] } : $conn->{errors}[-1];
}

sub yahc_conn_id            { $_[0]->{id}       }
sub yahc_conn_state         { $_[0]->{state}    }
sub yahc_conn_errors        { $_[0]->{errors}   }
sub yahc_conn_timeline      { $_[0]->{timeline} }
sub yahc_conn_request       { $_[0]->{request}  }
sub yahc_conn_response      { $_[0]->{response} }
sub yahc_conn_attempt       { $_[0]->{attempt}  }
sub yahc_conn_attempts_left { $_[0]->{attempt} > $_[0]->{retries} ? 0 : $_[0]->{retries} - $_[0]->{attempt} + 1 }

sub yahc_conn_target {
    my $target = $_[0]->{selected_target};
    return unless $target && scalar @{ $target };
    my ($host, $ip, $port) = @{ $target };
    return ($host || $ip) . ($port ne '80' && $port ne '443' ? ":$port" : '');
}

sub yahc_conn_url {
    my $target = $_[0]->{selected_target};
    my $request = $_[0]->{request};
    return unless $target && @{ $target };

    my ($host, $ip, $port, $scheme) = @{ $target };
    return "$scheme://"
           . ($host || $ip)
           . ($port ne '80' && $port ne '443' ? ":$port" : '')
           . ($request->{path} || "/")
           . (defined $request->{query_string} ? ("?" . $request->{query_string}) : "");
}

sub yahc_conn_user_data {
    my $conn = shift;
    $conn->{user_data} = $_[0] if @_;
    return $conn->{user_data};
}

################################################################################
# Internals
################################################################################

sub _run {
    my ($self, $how, $until_state, @cs) = @_;
    die "YAHC: storage object is destroyed\n" unless $self->{storage};
    die "YAHC: reentering run\n" if $self->{loop}->depth;

    if ($self->{pid} != $$) {
        _log_message('YAHC: reinitializing event loop after forking') if $self->{debug};
        $self->{pid} = $$;
        $self->{loop}->loop_fork;

        my $active_connections = grep { $$ != $_->{pid} } values %{ $self->{connections} };
        warn "YAHC has $active_connections active connections after a fork, consider dropping them!"
            if $active_connections;
    }

    if (defined $until_state) {
        my $until_state_str = _strstate($until_state);
        die "YAHC: unknown until_state $until_state\n" if $until_state_str =~ m/unknown/;

        my $is_all = (@cs == 0);
        my @connections = $is_all ? values %{ $self->{connections} }
                                  : map { $self->{connections}{$_} || () }
                                    map { ref($_) eq 'HASH' ? $_->{id} : $_ } @cs;

        $self->{stop_condition} = {
            all             => $is_all,
            expected_state  => $until_state,
            connections     => { map { $_->{id} => 1 } grep { $_->{state} < $until_state } @connections },
        };
    } else {
        delete $self->{stop_condition};
    }

    my $loop = $self->{loop};
    $loop->now_update;

    if ($self->{debug}) {
        my $iterations = $loop->iteration;
        _log_message('YAHC: pid %d entering event loop%s', $$, ($until_state ? " with until state " . _strstate($until_state) : ''));
        $loop->run($how || 0);
        _log_message('YAHC: pid %d exited from event loop after %d iterations', $$, $loop->iteration - $iterations);
    } else {
        $loop->run($how || 0);
    }
}

sub _check_stop_condition {
    my ($self, $conn) = @_;
    my $stop_condition = $self->{stop_condition};
    return if !$stop_condition || $conn->{state} < $stop_condition->{expected_state};

    delete $stop_condition->{connections}{$conn->{id}};
    my $awaiting_connections = scalar keys %{ $stop_condition->{connections} };
    my $expected_state = $stop_condition->{expected_state};

    if ($awaiting_connections == 0) {
        $self->break(sprintf("until state '%s' is reached", _strstate($expected_state)));
        return 1;
    }

    _log_message("YAHC: still have %d connections awaiting state '%s'",
                 $awaiting_connections, _strstate($expected_state)) if $self->{debug};
}

################################################################################
# IO routines
################################################################################

sub _set_init_state {
    my ($self, $conn_id) = @_;

    my $conn = $self->{connections}{$conn_id}  or die "YAHC: unknown connection id $conn_id\n";

    $conn->{response} = { status => 0 };
    $conn->{state} = YAHC::State::INITIALIZED();
    _register_in_timeline($conn, "new state %s", _strstate($conn->{state})) if exists $conn->{debug_or_timeline};
    _call_state_callback($self, $conn, 'init_callback') if exists $conn->{has_init_callback};

    _close_or_cache_socket($self, $conn, 1); # force connection close if any (likely not)
    my $watchers = _delete_watchers_but_lifetime_timer($self, $conn_id); # implicit stop of all watchers

    return _set_user_action_state($self, $conn_id, YAHC::Error::RETRY_LIMIT(), "retries limit reached")
        if $conn->{attempt} > $conn->{retries};

    # don't move attempt increment before boundary check !!!
    # otherwise we can get off-by-one error in yahc_conn_attempts_left
    my $attempt = ++$conn->{attempt};
    if ($attempt > 1 && exists $conn->{request}{_backoff}) {
        my $backoff_delay = eval { $conn->{request}{_backoff}->($conn) };
        if (my $error = $@) {
            return _set_user_action_state($self, $conn_id, YAHC::Error::CALLBACK_ERROR() | YAHC::Error::TERMINAL_ERROR(),
                "exception in backoff callback (close connection): $error");
        };

        if ($backoff_delay) {
            $self->{loop}->now_update;
            _register_in_timeline($conn, "setting backoff_timer to %.3fs", $backoff_delay) if exists $conn->{debug_or_timeline};
            $watchers->{backoff_timer} = $self->{loop}->timer($backoff_delay, 0, _get_safe_wrapper($self, $conn, sub {
                _register_in_timeline($conn, "backoff timer of %.3fs expired, time for new attempt", $backoff_delay) if exists $conn->{debug_or_timeline};
                _set_init_state($self, $conn_id) if _init_helper($self, $conn_id) == 1;
            }));
            return;
        }
    }

    if (_init_helper($self, $conn_id) == 1) {
        _register_in_timeline($conn, "do attempt on next EV iteration, (iteration=%d)", $self->{loop}->iteration)
            if exists $conn->{debug_or_timeline};

        # from EV docs:
        # idle watcher call the callback when there are no other pending
        # watchers of the same or higher priority. The idle watchers are
        # being called once per event loop iteration - until stopped.
        #
        # so, what we do is we start idle watcher with priority 1 which is
        # higher then 0 used by all IO watchers. As result, the callback
        # will be called at the end of this iteration. And others if neccessary.

        my $retry_watcher = $watchers->{retry} ||= $self->{loop}->idle_ns(_get_safe_wrapper($self, $conn, sub {
            shift->stop; # stop this watcher, _set_init_state will start if neccessary
            _register_in_timeline($conn, "time for new attempt (iteration=%d)", $self->{loop}->iteration)
                if exists $conn->{debug_or_timeline};
            _set_init_state($self, $conn_id)
        }));

        $retry_watcher->priority(1);
        $retry_watcher->start;
    };
}

sub _init_helper {
    my ($self, $conn_id) = @_;

    my $conn = $self->{connections}{$conn_id}  or die "YAHC: unknown connection id $conn_id\n";
    my $watchers = $self->{watchers}{$conn_id} or die "YAHC: no watchers for connection id $conn_id\n";

    my $request = $conn->{request};

    $self->{loop}->now_update; # update time for timers
    _set_until_state_timer($self, $conn_id, 'request_timeout', YAHC::State::USER_ACTION(), YAHC::Error::TIMEOUT() | YAHC::Error::REQUEST_TIMEOUT())
        if $request->{request_timeout};
    _set_until_state_timer($self, $conn_id, 'connect_timeout', YAHC::State::CONNECTED(),   YAHC::Error::TIMEOUT() | YAHC::Error::CONNECT_TIMEOUT())
        if $request->{connect_timeout};
    _set_until_state_timer($self, $conn_id, 'drain_timeout',   YAHC::State::READING(),     YAHC::Error::TIMEOUT() | YAHC::Error::DRAIN_TIMEOUT())
        if $request->{drain_timeout};

    eval {
        my ($host, $ip, $port, $scheme) = _get_next_target($conn);
        _register_in_timeline($conn, "Target $scheme://$host:$port ($ip:$port) chosen for attempt #%d", $conn->{attempt})
            if exists $conn->{debug_or_timeline};

        my $sock;
        if (my $socket_cache = $request->{_socket_cache}) {
            $sock = $socket_cache->(YAHC::SocketCache::GET(), $conn);
        }

        if (defined $sock) {
            _register_in_timeline($conn, "reuse socket") if $conn->{debug_or_timeline};
            $watchers->{_fh} = $sock;
            $watchers->{io} = $self->{loop}->io($sock, EV::WRITE, sub {});
            _set_write_state($self, $conn_id);
        } else {
            _register_in_timeline($conn, "build new socket") if $conn->{debug_or_timeline};
            $sock = _build_socket_and_connect($ip, $port);
            _set_connecting_state($self, $conn_id, $sock);
        }

        1;
    } or do {
        my $error = $@ || 'zombie error';
        $error =~ s/\s+$//o;
        yahc_conn_register_error($conn, YAHC::Error::CONNECT_ERROR(), "connection attempt %d failed: %s", $conn->{attempt}, $error);
        return 1;
    };

    return 0;
}

sub _set_connecting_state {
    my ($self, $conn_id, $sock) = @_;

    my $conn = $self->{connections}{$conn_id}  or die "YAHC: unknown connection id $conn_id\n";
    my $watchers = $self->{watchers}{$conn_id} or die "YAHC: no watchers for connection id $conn_id\n";

    $conn->{state} = YAHC::State::CONNECTING();
    _register_in_timeline($conn, "new state %s", _strstate($conn->{state})) if exists $conn->{debug_or_timeline};
    _call_state_callback($self, $conn, 'connecting_callback') if exists $conn->{has_connecting_callback};

    my $connecting_cb = _get_safe_wrapper($self, $conn, sub {
        my $sockopt = getsockopt($sock, SOL_SOCKET, SO_ERROR);
        if (!$sockopt) {
            yahc_conn_register_error($conn, YAHC::Error::CONNECT_ERROR(), "Failed to do getsockopt(): '%s' errno=%d", "$!", $!+0);
            _set_init_state($self, $conn_id);
            return;
        }

        if (my $err = unpack("L", $sockopt)) {
            my $strerror = POSIX::strerror($err) || '';
            yahc_conn_register_error($conn, YAHC::Error::CONNECT_ERROR(), "Failed to connect: $strerror");
            _set_init_state($self, $conn_id);
            return;
        }

        _set_connected_state($self, $conn_id);
    });

    $watchers->{_fh} = $sock;
    $watchers->{io} = $self->{loop}->io($sock, EV::WRITE, $connecting_cb);
    _check_stop_condition($self, $conn) if exists $self->{stop_condition};
}

sub _set_connected_state {
    my ($self, $conn_id) = @_;

    my $conn = $self->{connections}{$conn_id}  or die "YAHC: unknown connection id $conn_id\n";
    my $watchers = $self->{watchers}{$conn_id} or die "YAHC: no watchers for connection id $conn_id\n";

    $conn->{state} = YAHC::State::CONNECTED();
    _register_in_timeline($conn, "new state %s", _strstate($conn->{state})) if exists $conn->{debug_or_timeline};
    _call_state_callback($self, $conn, 'connected_callback') if exists $conn->{has_connected_callback};

    my $connected_cb = _get_safe_wrapper($self, $conn, sub {
        if ($conn->{is_ssl}) {
            _set_ssl_handshake_state($self, $conn_id);
        } else {
            _set_write_state($self, $conn_id);
        }
    });

    #$watcher->events(EV::WRITE);
    $watchers->{io}->cb($connected_cb);
    _check_stop_condition($self, $conn) if exists $self->{stop_condition};
}

sub _set_ssl_handshake_state {
    my ($self, $conn_id) = @_;

    my $conn = $self->{connections}{$conn_id}  or die "YAHC: unknown connection id $conn_id\n";
    my $watchers = $self->{watchers}{$conn_id} or die "YAHC: no watchers for connection id $conn_id\n";

    $conn->{state} = YAHC::State::SSL_HANDSHAKE();
    _register_in_timeline($conn, "new state %s", _strstate($conn->{state})) if exists $conn->{debug_or_timeline};
    #_call_state_callback($self, $conn, 'writing_callback') if $conn->{has_writing_callback}; TODO

    my $fh = $watchers->{_fh};
    my $hostname = $conn->{selected_target}[0];

    my %options = (
        SSL_verifycn_name => $hostname,
        IO::Socket::SSL->can_client_sni ? ( SSL_hostname => $hostname ) : (),
        %{ $conn->{request}{ssl_options} || {} },
    );

    if ($conn->{debug_or_timeline}) {
        my $options_msg = join(', ', map { "$_=" . ($options{$_} || '') } keys %options);
        _register_in_timeline($conn, "start SSL handshake with options: $options_msg");
    }

    if (!IO::Socket::SSL->start_SSL($fh, %options, SSL_startHandshake => 0)) {
        return _set_user_action_state($self, $conn_id, YAHC::Error::SSL_ERROR() | YAHC::Error::TERMINAL_ERROR(),
            sprintf("failed to start SSL session: %s", _format_ssl_error()));
    }

    my $handshake_cb = _get_safe_wrapper($self, $conn, sub {
        my $w = shift;
        if ($fh->connect_SSL) {
            _register_in_timeline($conn, "SSL handshake successfully completed") if exists $conn->{debug_or_timeline};
            return _set_write_state($self, $conn_id);
        }

        if ($! == EWOULDBLOCK) {
            return $w->events(EV::READ)  if $IO::Socket::SSL::SSL_ERROR == SSL_WANT_READ;
            return $w->events(EV::WRITE) if $IO::Socket::SSL::SSL_ERROR == SSL_WANT_WRITE;
        }

        yahc_conn_register_error($conn, YAHC::Error::SSL_ERROR(), "Failed to complete SSL handshake: %s", _format_ssl_error());
        _set_init_state($self, $conn_id);
    });

    my $watcher = $watchers->{io};
    $watcher->cb($handshake_cb);
    $watcher->events(EV::WRITE | EV::READ);
    _check_stop_condition($self, $conn) if exists $self->{stop_condition};
}

sub _set_write_state {
    my ($self, $conn_id) = @_;

    my $conn = $self->{connections}{$conn_id}  or die "YAHC: unknown connection id $conn_id\n";
    my $watchers = $self->{watchers}{$conn_id} or die "YAHC: no watchers for connection id $conn_id\n";

    $conn->{state} = YAHC::State::WRITING();
    _register_in_timeline($conn, "new state %s", _strstate($conn->{state})) if exists $conn->{debug_or_timeline};
    _call_state_callback($self, $conn, 'writing_callback') if exists $conn->{has_writing_callback};

    my $fh = $watchers->{_fh};
    my $buf = _build_http_message($conn);
    my $length = length($buf);

    warn "YAHC: HTTP message has UTF8 flag set! This will result in poor performance, see docs for details!"
        if utf8::is_utf8($buf);

    _register_in_timeline($conn, "writing body of %d bytes\n%s", $length, ($length > 1024? substr($buf, 0, 1024) . '... (cut to 1024 bytes)' : $buf))
        if exists $conn->{debug_or_timeline};

    my $write_cb = _get_safe_wrapper($self, $conn, sub {
        my $w = shift;
        my $wlen = syswrite($fh, $buf, $length);

        if (!defined $wlen) {
            if ($conn->{is_ssl}) {
                if ($! == EWOULDBLOCK) {
                    return $w->events(EV::READ)  if $IO::Socket::SSL::SSL_ERROR == SSL_WANT_READ;
                    return $w->events(EV::WRITE) if $IO::Socket::SSL::SSL_ERROR == SSL_WANT_WRITE;
                }

                yahc_conn_register_error($conn, YAHC::Error::WRITE_ERROR() | YAHC::Error::SSL_ERROR(), "Failed to send HTTPS data: %s", _format_ssl_error());
                return _set_init_state($self, $conn_id);
            }

            return if $! == EWOULDBLOCK || $! == EINTR || $! == EAGAIN;
            yahc_conn_register_error($conn, YAHC::Error::WRITE_ERROR(), "Failed to send HTTP data: '%s' errno=%d", "$!", $!+0);
            _set_init_state($self, $conn_id);
        } elsif ($wlen == 0) {
            yahc_conn_register_error($conn, YAHC::Error::WRITE_ERROR(), "syswrite returned 0");
            _set_init_state($self, $conn_id);
        } else {
            substr($buf, 0, $wlen, '');
            $length -= $wlen;
            _set_read_state($self, $conn_id) if $length == 0;
        }
    });

    my $watcher = $watchers->{io};
    $watcher->cb($write_cb);
    $watcher->events(EV::WRITE);
    _check_stop_condition($self, $conn) if exists $self->{stop_condition};
}

sub _set_read_state {
    my ($self, $conn_id) = @_;

    my $conn = $self->{connections}{$conn_id}  or die "YAHC: unknown connection id $conn_id\n";
    my $watchers = $self->{watchers}{$conn_id} or die "YAHC: no watchers for connection id $conn_id\n";

    $conn->{state} = YAHC::State::READING();
    _register_in_timeline($conn, "new state %s", _strstate($conn->{state})) if exists $conn->{debug_or_timeline};
    _call_state_callback($self, $conn, 'reading_callback') if exists $conn->{has_reading_callback};

    my $buf = '';
    my $neck_pos = 0;
    my $decapitated = 0;
    my $content_length = 0;
    my $is_chunked = 0;
    my $fh = $watchers->{_fh};
    my $chunk_size = 0;
    my $body = ''; # used for chunked encoding

    my $read_cb = _get_safe_wrapper($self, $conn, sub {
        my $w = shift;
        my $rlen = sysread($fh, my $b = '', TCP_READ_CHUNK);
        if (!defined $rlen) {
            if ($conn->{is_ssl}) {
                if ($! == EWOULDBLOCK) {
                    return $w->events(EV::READ)  if $IO::Socket::SSL::SSL_ERROR == SSL_WANT_READ;
                    return $w->events(EV::WRITE) if $IO::Socket::SSL::SSL_ERROR == SSL_WANT_WRITE;
                }

                yahc_conn_register_error($conn, YAHC::Error::READ_ERROR() | YAHC::Error::SSL_ERROR(), "Failed to receive HTTPS data: %s", _format_ssl_error());
                return _set_init_state($self, $conn_id);
            }

            return if $! == EWOULDBLOCK || $! == EINTR || $! == EAGAIN;
            yahc_conn_register_error($conn, YAHC::Error::READ_ERROR(), "Failed to receive HTTP data: '%s' errno=%d", "$!", $!+0);
            _set_init_state($self, $conn_id);
        } elsif ($rlen == 0) {
            if ($content_length > 0) {
                yahc_conn_register_error($conn, YAHC::Error::READ_ERROR(), "Premature EOF, expect %d bytes more", $content_length - length($buf));
            } else {
                yahc_conn_register_error($conn, YAHC::Error::READ_ERROR(), "Premature EOF");
            }
            _set_init_state($self, $conn_id);
        } else {
            $buf .= $b;
            if (!$decapitated && ($neck_pos = index($buf, "${CRLF}${CRLF}")) > 0) {
                my $headers = _parse_http_headers($conn, substr($buf, 0, $neck_pos, '')); # $headers are always defined but might be empty, maybe fix later
                $is_chunked = ($headers->{'Transfer-Encoding'} || '') eq 'chunked';

                if ($is_chunked && exists $headers->{'Trailer'}) {
                    _set_user_action_state($self, $conn_id, YAHC::Error::RESPONSE_ERROR(), "Chunked HTTP response with Trailer header");
                    return;
                } elsif (!$is_chunked && !exists $headers->{'Content-Length'}) {
                    _set_user_action_state($self, $conn_id, YAHC::Error::RESPONSE_ERROR(), "HTTP reponse without Content-Length");
                    return;
                }

                $decapitated = 1;
                $content_length = $headers->{'Content-Length'};
                substr($buf, 0, 4, ''); # 4 = length("$CRLF$CRLF")
            }

            if ($decapitated && $is_chunked) {
                # in order to get the smallest chunk size we need
                # at least 4 bytes (2xCLRF), and there *MUST* be
                # last chunk which is at least 5 bytes (0\r\n\r\n)
                # so we can safely ignore $bufs that have less than 5 bytes
                while (length($buf) > ($chunk_size + 4)) {
                    my $neck_pos = index($buf, ${CRLF});
                    if ($neck_pos > 0) {
                        # http://www.w3.org/Protocols/rfc2616/rfc2616-sec3.html
                        # All HTTP/1.1 applications MUST be able to receive and
                        # decode the "chunked" transfer-coding, and MUST ignore
                        # chunk-extension extensions they do not understand.
                        my ($s) = split(';', substr($buf, 0, $neck_pos), 1);
                        $chunk_size = hex($s);

                        _register_in_timeline($conn, "parsing chunk of size $chunk_size bytes") if exists $conn->{debug_or_timeline};
                        if ($chunk_size == 0) { # end with, but as soon as we see 0\r\n\r\n we just mark it as done
                            $conn->{response}{body} = $body;
                            _set_user_action_state($self, $conn_id);
                            return;
                        } else {
                            if (length($buf) >= $chunk_size + $neck_pos + 2 + 2) {
                                $body .= substr($buf, $neck_pos + 2, $chunk_size);
                                substr($buf, 0, $neck_pos + 2 + $chunk_size + 2, '');
                                $chunk_size = 0;
                            } else {
                                last; # dont have enough data in this pass, wait for one more read
                            }
                        }
                    } else {
                        last if $neck_pos < 0 && $chunk_size == 0; # in case we couldnt get the chunk size in one go, we must concat until we have something
                        _set_user_action_state($self, $conn_id, YAHC::Error::RESPONSE_ERROR(), "error processing chunked data, couldnt find CLRF[index:$neck_pos] in buf");
                        return;
                    }
                }
            } elsif ($decapitated && ($conn->{request}{method} && $conn->{request}{method} eq 'HEAD')) {
                _set_user_action_state($self, $conn_id); # We are done reading headers for a HEAD request
            } elsif ($decapitated && length($buf) >= $content_length) {
                $conn->{response}{body} = (length($buf) > $content_length ? substr($buf, 0, $content_length) : $buf);
                _set_user_action_state($self, $conn_id);
            }
        }
    });

    my $watcher = $watchers->{io};
    $watcher->cb($read_cb);
    $watcher->events(EV::READ);
    _check_stop_condition($self, $conn) if exists $self->{stop_condition};
}

sub _set_user_action_state {
    my ($self, $conn_id, $error, $strerror) = @_;
    $error ||= YAHC::Error::NO_ERROR();
    $strerror ||= '';

    # this state may be used in critical places,
    # so it should *NEVER* throw exception
    my $conn = $self->{connections}{$conn_id}
      or warn "YAHC: try to _set_user_action_state() for unknown connection $conn_id",
        return;

    $conn->{state} = YAHC::State::USER_ACTION();
    _register_in_timeline($conn, "new state %s", _strstate($conn->{state})) if exists $conn->{debug_or_timeline};
    yahc_conn_register_error($conn, $error, $strerror) if $error != YAHC::Error::NO_ERROR;

    _close_or_cache_socket($self, $conn, $error != YAHC::Error::NO_ERROR);
    return _set_completed_state($self, $conn_id) unless exists $conn->{has_callback};

    eval {
        _register_in_timeline($conn, "call callback%s", $error ? " error=$error, strerror='$strerror'" : '') if exists $conn->{debug_or_timeline};
        my $cb = $self->{callbacks}{$conn_id}{callback};
        $cb->($conn, $error, $strerror);
        1;
    } or do {
        my $error = $@ || 'zombie error';
        yahc_conn_register_error($conn, YAHC::Error::CALLBACK_ERROR() | YAHC::Error::TERMINAL_ERROR(), "Exception in user action callback (close connection): $error");
        $self->{state} = YAHC::State::COMPLETED();
    };

    $self->{loop}->now_update;

    my $state = $conn->{state};
    if (yahc_terminal_error($error)) {
        yahc_conn_register_error($conn, YAHC::Error::CALLBACK_ERROR() | YAHC::Error::TERMINAL_ERROR(), "ignoring changed state due to terminal error")
            unless $state == YAHC::State::USER_ACTION() || $state == YAHC::State::COMPLETED();
        _set_completed_state($self, $conn_id, 1);
        return
    }

    _register_in_timeline($conn, "after invoking callback state is %s", _strstate($state)) if exists $conn->{debug_or_timeline};

    if ($state == YAHC::State::INITIALIZED()) {
        _set_init_state($self, $conn_id);
    } elsif ($state == YAHC::State::USER_ACTION() || $state == YAHC::State::COMPLETED()) {
        _set_completed_state($self, $conn_id);
    } else {
        yahc_conn_register_error($conn, YAHC::Error::CALLBACK_ERROR() | YAHC::Error::TERMINAL_ERROR(), "callback set unsupported state");
        _set_completed_state($self, $conn_id);
    }
}

sub _set_completed_state {
    my ($self, $conn_id, $force_socket_close) = @_;

    # this's a terminal state,
    # so setting this state should *NEVER* fail
    delete $self->{callbacks}{$conn_id};
    my $conn = delete $self->{connections}{$conn_id};

    if (!defined $conn) {
        delete($self->{watchers}{$conn_id}), # implicit stop of all watchers
        return;
    }

    $conn->{state} = YAHC::State::COMPLETED();
    _register_in_timeline($conn, "new state %s", _strstate($conn->{state})) if exists $conn->{debug_or_timeline};

    _close_or_cache_socket($self, $conn, $force_socket_close);
    delete $self->{watchers}{$conn_id}; # implicit stop of all watchers

    _check_stop_condition($self, $conn) if exists $self->{stop_condition};
}

sub _build_socket_and_connect {
    my ($ip, $port) = @_;

    my $sock;
    socket($sock, PF_INET, SOCK_STREAM, 0)
        or die sprintf("Failed to construct TCP socket: '%s' errno=%d\n", "$!", $!+0);

    my $flags = fcntl($sock, F_GETFL, 0) or die sprintf("Failed to get fcntl F_GETFL flag: '%s' errno=%d\n", "$!", $!+0);
    fcntl($sock, F_SETFL, $flags | O_NONBLOCK) or die sprintf("Failed to set fcntl O_NONBLOCK flag: '%s' errno=%d\n", "$!", $!+0);

    my $ip_addr = inet_aton($ip) or die "Invalid IP address";
    my $addr = pack_sockaddr_in($port, $ip_addr);
    if (!connect($sock, $addr) && $! != EINPROGRESS) {
        die sprintf("Failed to connect: '%s' errno=%d\n", "$!", $!+0);
    }

    return $sock;
}

sub _get_next_target {
    my $conn = shift;
    my ($host, $ip, $port, $scheme) = $conn->{request}{_target}->($conn);

    # TODO STATE_RESOLVE_DNS
    ($host, $port) = ($1, $2) if !$port && $host =~ m/^(.+):([0-9]+)$/o;
    $ip = $host if !$ip && $host =~ m/^[0-9]+\.[0-9]+\.[0-9]+\.[0-9]+$/o;
    $ip ||= inet_ntoa(gethostbyname($host) or die "Failed to resolve $host\n");
    $scheme ||= $conn->{request}{scheme} || 'http';
    $port   ||= $conn->{request}{port} || ($scheme eq 'https' ? 443 : 80);

    $conn->{is_ssl} = $scheme eq 'https';
    return @{ $conn->{selected_target} = [ $host, $ip, $port, $scheme ] };
}

# this and following functions are used in terminal state
# so they should *NEVER* fail
sub _close_or_cache_socket {
    my ($self, $conn, $force_close) = @_;
    my $watchers = $self->{watchers}{$conn->{id}} or return;
    my $fh = delete $watchers->{_fh} or return;
    delete $watchers->{io}; # implicit stop

    my $socket_cache = $conn->{request}{_socket_cache};

    # Stolen from Hijk. Thanks guys!!!
    # We always close connections for 1.0 because some servers LIE
    # and say that they're 1.0 but don't close the connection on
    # us! An example of this. Test::HTTP::Server (used by the
    # ShardedKV::Storage::Rest tests) is an example of such a
    # server. In either case we can't cache a connection for a 1.0
    # server anyway, so BEGONE!

    if (   $force_close
        || !defined $socket_cache
        || (($conn->{request}{proto} || '') eq 'HTTP/1.0')
        || (($conn->{response}{proto} || '') eq 'HTTP/1.0')
        || (($conn->{response}{head}{Connection} || '') eq 'close'))
    {
        _register_in_timeline($conn, "drop socket") if $conn->{debug_or_timeline};
        close($fh) if ref($fh) eq 'GLOB'; # checking ref to avoid exception
        return;
    }

    _register_in_timeline($conn, "storing socket for later use") if $conn->{debug_or_timeline};
    eval { $socket_cache->(YAHC::SocketCache::STORE(), $conn, $fh); 1; } or do {
        yahc_conn_register_error($conn, YAHC::Error::CALLBACK_ERROR(), "Exception in socket_cache callback (ignore error): $@");
    };
}

sub yahc_conn_socket_cache_id {
    my $conn = shift;
    return unless defined $conn;
    my ($host, undef, $port, $scheme) = @{ $conn->{selected_target} || [] };
    return unless $host && $port && $scheme;
    # Use $; so we can use the $socket_cache->{$$, $host, $port} idiom to access the cache.
    return join($;, $$, $host, $port, $scheme);
}

################################################################################
# Timers
################################################################################

sub _set_until_state_timer {
    my ($self, $conn_id, $timeout_name, $state, $error_to_report) = @_;

    my $timer_name = $timeout_name . '_timer';
    my $conn = $self->{connections}{$conn_id}  or die "YAHC: unknown connection id $conn_id\n";
    my $watchers = $self->{watchers}{$conn_id} or die "YAHC: no watchers for connection id $conn_id\n";

    delete $watchers->{$timer_name}; # implicit stop
    my $timeout = $conn->{request}{$timeout_name};
    return unless $timeout;

    my $timer_cb = sub { # there is nothing what can throw exception
        if ($conn->{state} < $state) {
            yahc_conn_register_error($conn, $error_to_report, "$timeout_name of %.3fs expired", $timeout);
            _set_init_state($self, $conn_id);
        } else {
            _register_in_timeline($conn, "delete $timer_name") if exists $conn->{debug_or_timeline};
        }
    };

    _register_in_timeline($conn, "setting $timeout_name to %.3fs", $timeout) if exists $conn->{debug_or_timeline};

    # caller should call now_update
    my $w = $watchers->{$timer_name} = $self->{loop}->timer_ns($timeout, 0, $timer_cb);
    $w->priority(2); # set highest priority
    $w->start;
}

sub _set_lifetime_timer {
    my ($self, $conn_id) = @_;

    my $conn = $self->{connections}{$conn_id}  or die "YAHC: unknown connection id $conn_id\n";
    my $watchers = $self->{watchers}{$conn_id} or die "YAHC: no watchers for connection id $conn_id\n";

    delete $watchers->{lifetime_timer}; # implicit stop
    my $timeout = $conn->{request}{lifetime_timeout};
    return unless $timeout;

    _register_in_timeline($conn, "setting lifetime timer to %.3fs", $timeout) if exists $conn->{debug_or_timeline};

    $self->{loop}->now_update;
    my $w = $watchers->{lifetime_timer} = $self->{loop}->timer_ns($timeout, 0, sub {
        _set_user_action_state($self, $conn_id, YAHC::Error::TIMEOUT() | YAHC::Error::LIFETIME_TIMEOUT() | YAHC::Error::TERMINAL_ERROR(),
            sprintf("lifetime_timeout of %.3fs expired", $timeout)) if $conn->{state} < YAHC::State::COMPLETED();
    });

    $w->priority(2); # set highest priority
    $w->start;
}

################################################################################
# HTTP functions
################################################################################

# copy-paste from Hijk
sub _build_http_message {
    my $conn = shift;
    my $request = $conn->{request};
    my $path_and_qs = ($request->{path} || "/") . (defined $request->{query_string} ? ("?" . $request->{query_string}) : "");
    my $has_host = 0;

    return join(
        $CRLF,
        ($request->{method} || "GET") . " $path_and_qs " . ($request->{protocol} || "HTTP/1.1"),
        defined($request->{body}) ? ("Content-Length: " . length($request->{body})) : (),
        defined($request->{head}) && @{ $request->{head} } ? (
            map {
                $has_host ||= lc($request->{head}[2*$_]) eq 'host';
                $request->{head}[2*$_] . ": " . $request->{head}[2*$_+1]
            } 0..$#{$request->{head}}/2
        ) : (),
        !$has_host ? ("Host: " . $conn->{selected_target}[0]) : (),
        "",
        defined($request->{body}) ? $request->{body} : ""
    );
}

sub _parse_http_headers {
    my $conn = shift;
    my $proto       = substr($_[0], 0, 8);
    my $status_code = substr($_[0], 9, 3);
    substr($_[0], 0, index($_[0], $CRLF) + 2, ''); # 2 = length($CRLF)

    my %headers;
    for (split /${CRLF}/o, $_[0]) {
        my ($key, $value) = split(/: /, $_, 2);
        $headers{$key} = $value;
    }

    $conn->{response} = {
        proto  => $proto,
        status => $status_code,
        head   => \%headers,
    };

    if ($conn->{debug_or_timeline}) {
        my $headers_str = join(' ', map { "$_='$headers{$_}'" } keys %headers);
        _register_in_timeline($conn, "headers parsed: $status_code $proto headers=$headers_str");
    }

    return \%headers;
}

################################################################################
# Helpers
################################################################################

sub _delete_watchers_but_lifetime_timer {
    my ($self, $conn_id) = @_;

    my $watchers = $self->{watchers}{$conn_id};
    if (defined $watchers && (my $w = $watchers->{lifetime_timer})) {
        return $self->{watchers}{$conn_id} = { lifetime_timer => $w };
    }

    return $self->{watchers}{$conn_id} = {};
}

sub _wrap_host {
    my ($value) = @_;
    my $ref = ref($value);

    return sub { $value } if $ref eq '';
    return $value         if $ref eq 'CODE';

    my $idx = 0;
    return sub { $value->[$idx++ % @$value]; }
        if $ref eq 'ARRAY' && @$value > 0;

    die "YAHC: unsupported host format\n";
}

sub _wrap_backoff {
    my ($value) = @_;
    my $ref = ref($value);

    return sub { $value } if $ref eq '';
    return $value         if $ref eq 'CODE';

    die "YAHC: unsupported backoff format\n";
}

sub _wrap_socket_cache {
    my ($value) = @_;
    my $ref = ref($value);

    return $value if $ref eq 'CODE';
    return sub {
        my ($operation, $conn, $sock) = @_;
        if ($operation == YAHC::SocketCache::GET()) {
            my $socket_cache_id = yahc_conn_socket_cache_id($conn) or return;
            return delete $value->{$socket_cache_id};
        }

        if ($operation == YAHC::SocketCache::STORE()) {
            my $socket_cache_id = yahc_conn_socket_cache_id($conn) or return;
            close(delete $value->{$socket_cache_id}) if exists $value->{$socket_cache_id};
            $value->{$socket_cache_id} = $sock;
            return;
        }
    } if $ref eq 'HASH';

    die "YAHC: unsupported socket_cache format\n";
}

sub _call_state_callback {
    my ($self, $conn, $cb_name) = @_;
    my $cb = $self->{callbacks}{$conn->{id}}{$cb_name};
    return unless $cb;

    _register_in_timeline($conn, "calling $cb_name callback") if exists $conn->{debug_or_timeline};

    eval {
        $cb->($conn);
        1;
    } or do {
        my $error = $@ || 'zombie error';
        yahc_conn_register_error($conn, YAHC::Error::CALLBACK_ERROR(), "exception in state callback (ignore error): $error");
    };

    $self->{loop}->now_update;
}

sub _get_safe_wrapper {
    my ($self, $conn, $sub) = @_;
    return sub { eval {
        $sub->(@_);
        1;
    } or do {
        my $error = $@ || 'zombie error';
        _set_user_action_state($self, $conn->{id}, YAHC::Error::INTERNAL_ERROR() | YAHC::Error::TERMINAL_ERROR(),
            "exception in internal callback: $error");
    }};
}

sub _register_in_timeline {
    my ($conn, $format, @arguments) = @_;
    my $event = sprintf("$format", @arguments);
    _log_message("YAHC connection '%s': %s", $conn->{id}, $event) if exists $conn->{debug};
    push @{ $conn->{timeline} ||= [] }, [ $event, $conn->{state}, Time::HiRes::time ] if exists $conn->{keep_timeline};
}

sub yahc_conn_register_error {
    my ($conn, $error, $format, @arguments) = @_;
    my $strerror = sprintf("$format", @arguments);
    _register_in_timeline($conn, "strerror='$strerror' error=$error") if exists $conn->{debug_or_timeline};
    push @{ $conn->{errors} ||= [] }, [ $error, $strerror, [ @{ $conn->{selected_target} } ], Time::HiRes::time, $conn->{attempt} ];
}

sub _strstate {
    my $state = shift;
    return 'STATE_INIT'         if $state eq YAHC::State::INITIALIZED();
    return 'STATE_RESOLVE_DNS'  if $state eq YAHC::State::RESOLVE_DNS();
    return 'STATE_CONNECTING'   if $state eq YAHC::State::CONNECTING();
    return 'STATE_CONNECTED'    if $state eq YAHC::State::CONNECTED();
    return 'STATE_WRITING'      if $state eq YAHC::State::WRITING();
    return 'STATE_READING'      if $state eq YAHC::State::READING();
    return 'STATE_SSL_HANDSHAKE'if $state eq YAHC::State::SSL_HANDSHAKE();
    return 'STATE_USER_ACTION'  if $state eq YAHC::State::USER_ACTION();
    return 'STATE_COMPLETED'    if $state eq YAHC::State::COMPLETED();
    return "";
}

sub _log_message {
    my $format = shift;
    my $now = Time::HiRes::time;
    my ($sec, $ms) = split(/[.]/, $now);
    printf STDERR "[%s.%-5d] [$$] $format\n", POSIX::strftime('%F %T', localtime($now)), $ms || 0, @_;
}

sub _format_ssl_error { return sprintf("'%s' errno=%d ssl_error='%s' ssl_errno=%d", "$!", 0+$!, "$IO::Socket::SSL::SSL_ERROR", 0+$IO::Socket::SSL::SSL_ERROR); }

1;

__END__