gistfile1.pl

package Coro::Mysql::DBI;

use strict;
use warnings;

use DBI;
use Coro;
use Coro::Mysql;
use Hash::FieldHash qw(:all);

# our $ROOT_CLASS = 'DBIx::ContextualFetch';
our $ROOT_CLASS = 'DBI';

# MAX CONNECTION for same DB
our $MAX_CACHED_CONNECTIONS = 5;

# unblocking
our $unblock = 1;

# private
fieldhash my %coro_of;
my $conn_cache;

# connect & managed by Coro
sub connect {
    my $class = shift;
    my @info = @_;
    my $current = $Coro::current;
    my $dbh = $ROOT_CLASS->connect(@_);
    Coro::Mysql::unblock $dbh if $unblock;
    $coro_of{$dbh} = $current;
    $current->on_destroy(sub { $coro_of{$dbh} = undef });
    return $dbh;
}

# connect_cached by coro, but other coro create new connection
sub connect_cached {
    my $class = shift;
    my @info = @_;
    my $current = $Coro::current;
    my $cache = ($conn_cache ||= Coro::Mysql::DBI::ConnCache->new($MAX_CACHED_CONNECTIONS));
    my $key = _cache_key(@info);
    my $dbh = $cache->get($key);
    unless ($dbh) {
        # if connections of $key > $MAX_CACHED_CONNECTIONS, wait for other coro release $dbh
        my $lock_success = $cache->get_lock($key);
        if (!$lock_success) {
            $cache->wait_for_release($key);
            # retry find cached dbh
            $dbh = $cache->get($key);
        }
        unless ($dbh) {
            $dbh = $ROOT_CLASS->connect(@info);
            Coro::Mysql::unblock $dbh if $unblock;
            $cache->set($key, $dbh);
        }
    }
    $coro_of{$dbh} = $current;
    $current->on_destroy(sub {
        $coro_of{$dbh} = undef;
        $cache->release_lock($key);
    });
    return $dbh;
}

sub coro_of {
    my ($class, $dbh) = @_;
    return unless $dbh;
    $coro_of{$dbh};
}

# copy from DBI
sub _cache_key {
    my ($dsn, $user, $auth, $attr) = @_;
    $attr ||= {};
    my $key = do { local $^W;
        join "!\001", $dsn, $user, $auth, DBI::_concat_hash_sorted($attr, "=\001", ",\001", 0, 0)
    };
}

1;

package Coro::Mysql::DBI::ConnCache;
use strict;
use warnings;
use Coro;
use Coro::Semaphore;

our $DEBUG;
my %CACHE;
my %LOCK;

sub new {
    my ($class, $max) = @_;
    bless {
        max => $max,
    }, $class;
}

sub get {
    my ($self, $key) = @_;
    my ($dbh) = $self->_select_usable_dbh($key);
    return $dbh;
}

sub wait_for_release {
    my ($self, $key) = @_;
    my $lock = ($LOCK{$key} ||= Coro::Semaphore->new($self->{max}));
    # warn $lock->count;
    $lock->down;
}

sub get_lock {
    my ($self, $key) = @_;
    my $lock = ($LOCK{$key} ||= Coro::Semaphore->new($self->{max}));
    # warn $lock->count;
    $lock->try;
}

sub release_lock {
    my ($self, $key) = @_;
    my $lock = ($LOCK{$key} ||= Coro::Semaphore->new($self->{max}));
    # warn $lock->count;
    $lock->up;
}

sub get_active_count {
    my ($self, $key) = @_;
    my $count = $self->select_all_dbh($key);
    return $count;
}

sub set {
    my ($self, $key, $dbh) = @_;
    my $cache = ($CACHE{$key} ||= []);
    push @{$cache}, $dbh;
    return 1;
}

sub _select_usable_dbh {
    my ($self, $key) = @_;
    my $cache = ($CACHE{$key} ||= []);
    my @usable;
    my $current = $Coro::current;
    # warn $current;
    my @pairs = map {
        [$_, Coro::Mysql::DBI->coro_of($_)]
    } @{$cache};

    # cached by current coro
    for my $pair (@pairs) {
        my ($dbh, $coro) = @{$pair};
        if ($coro && $coro == $current) {
            return $dbh;
        }
    }

    for my $pair (@pairs) {
        my ($dbh, $coro) = @{$pair};
        # used by other coro
        next if $coro;
        # reuse connection what is other coro used
        # TODO: ping
        # unless ($dbh && $dbh->FETCH('Active') && $dbh->ping) {
        unless ($dbh && $dbh->FETCH('Active')) {
            warn "dead $dbh" if $DEBUG;
            $dbh = undef;
            next;
        }
        push @usable, $dbh;
    }
    @{$cache} = grep { $_ } @{$cache};
    return @usable;
}

sub _select_all_dbh {
    my ($self, $key) = @_;
    my $cache = ($CACHE{$key} ||= []);
    return grep {$_} @{$cache};
}

1;