AnyEvent-DBI/DBI.pm

=head1 NAME

AnyEvent::DBI - asynchronous DBI access

=head1 SYNOPSIS

   use AnyEvent::DBI;

   my $cv = AnyEvent->condvar;

   my $dbh = new AnyEvent::DBI "DBI:SQLite:dbname=test.db", "", "";

   $dbh->exec ("select * from test where num=?", 10, sub {
      my ($dbh, $rows, $rv) = @_;

      $#_ or die "failure: $@";

      print "@$_\n"
         for @$rows;

      $cv->broadcast;
   });

   # asynchronously do sth. else here

   $cv->wait;

=head1 DESCRIPTION

This module is an L<AnyEvent> user, you need to make sure that you use and
run a supported event loop.

This module implements asynchronous DBI access by forking or executing
separate "DBI-Server" processes and sending them requests.

It means that you can run DBI requests in parallel to other tasks.

The overhead for very simple statements ("select 0") is somewhere
around 120% to 200% (dual/single core CPU) compared to an explicit
prepare_cached/execute/fetchrow_arrayref/finish combination.

=head2 ERROR HANDLING

This module defines a number of functions that accept a callback
argument. All callbacks used by this module get their AnyEvent::DBI handle
object passed as first argument.

If the request was successful, then there will be more arguments,
otherwise there will only be the C<$dbh> argument and C<$@> contains an
error message.

A convinient way to check whether an error occured is to check C<$#_> -
if that is true, then the function was successful, otherwise there was an
error.

=cut

package AnyEvent::DBI;

use strict qw(vars subs);
no warnings;

use Carp;
use Socket ();
use Scalar::Util ();
use Storable ();

use DBI ();

use AnyEvent ();
use AnyEvent::Util ();

use Errno ();
use Fcntl ();
use POSIX ();

our $VERSION = '2.0';

our $FD_MAX = eval { POSIX::sysconf (&POSIX::_SC_OPEN_MAX) - 1 } || 1023;

# this is the forked server code, could/should be bundled as it's own file

our $DBH;

sub req_open {
   my (undef, $dbi, $user, $pass, %attr) = @{+shift};

   $DBH = DBI->connect ($dbi, $user, $pass, \%attr) or die $DBI::errstr;

   [1, 1]
}

sub req_exec {
   my (undef, $st, @args) = @{+shift};
   my $sth = $DBH->prepare_cached ($st, undef, 1)
      or die [$DBI::errstr];

   my $rv = $sth->execute (@args)
      or die [$sth->errstr];

   [1, $sth->{NUM_OF_FIELDS} ? $sth->fetchall_arrayref : undef, $rv]
}

sub req_attr {
   my (undef, $attr_name, @attr_val) = @{+shift};

   $DBH->{$attr_name} = $attr_val[0]
      if @attr_val;

   [1, $DBH->{$attr_name}]
}

sub req_begin_work {
   [1, $DBH->begin_work or die [$DBI::errstr]]
}

sub req_commit {
   [1, $DBH->commit     or die [$DBI::errstr]]
}

sub req_rollback {
   [1, $DBH->rollback   or die [$DBI::errstr]]
}

sub req_func {
   my (undef, $arg_string, $function) = @{+shift};
   my @args = eval $arg_string;

   die "error evaling \$dbh->func() arg_string: $@"
      if $@;

   my $rc = $DBH->func (@args, $function);
   return [1, $rc, $DBI::err, $DBI::errstr];
}

sub serve_fh($$) {
   my ($fh, $version) = @_;

   if ($VERSION != $version) {
      syswrite $fh,
         pack "L/a*",
            Storable::freeze
               [undef, "AnyEvent::DBI version mismatch ($VERSION vs. $version)"];
      return;
   }

   eval {
      my $rbuf;

      while () {
         sysread $fh, $rbuf, 16384, length $rbuf
            or last;

         while () {
            my $len = unpack "L", $rbuf;

            # full request available?
            last unless $len && $len + 4 <= length $rbuf;

            my $req = Storable::thaw substr $rbuf, 4;
            substr $rbuf, 0, $len + 4, ""; # remove length + request

            my $wbuf = eval { pack "L/a*", Storable::freeze $req->[0]($req) };
            $wbuf = pack "L/a*", Storable::freeze [undef, ref $@ ? ("$@->[0]", $@->[1]) : ("$@", 1)]
               if $@;

            for (my $ofs = 0; $ofs < length $wbuf; ) {
               $ofs += (syswrite $fh, substr $wbuf, $ofs
                           or die "unable to write results");
            }
         }
      }
   };
}

sub serve_fd($$) {
   open my $fh, ">>&=$_[0]"
      or die "Couldn't open server file descriptor: $!";

   serve_fh $fh, $_[1];
}

=head2 METHODS

=over 4

=item $dbh = new AnyEvent::DBI $database, $user, $pass, [key => value]...

Returns a database handle for the given database. Each database handle
has an associated server process that executes statements in order. If
you want to run more than one statement in parallel, you need to create
additional database handles.

The advantage of this approach is that transactions work as state is
preserved.

Example:

   $dbh = new AnyEvent::DBI
             "DBI:mysql:test;mysql_read_default_file=/root/.my.cnf", "", "";

Additional key-value pairs can be used to adjust behaviour:

=over 4

=item on_error => $callback->($dbh, $filename, $line, $fatal)

When an error occurs, then this callback will be invoked. On entry, C<$@>
is set to the error message. C<$filename> and C<$line> is where the
original request was submitted.

If the fatal argument is true then the database connection is shut down
and your database handle became invalid. In addition to invoking the
C<on_error> callback, all of your queued request callbacks are called
without only the C<$dbh> argument.

If omitted, then C<die> will be called on any errors, fatal or not.

=item on_connect => $callback->($dbh[, $success])

If you supply an C<on_connect> callback, then this callback will be
invoked after the database connect attempt. If the connection succeeds,
C<$success> is true, otherwise it is missing and C<$@> contains the
C<$DBI::errstr>.

Regardless of whether C<on_connect> is supplied, connect errors will result in
C<on_error> being called. However, if no C<on_connect> callback is supplied, then
connection errors are considered fatal. The client will C<die> and the C<on_error>
callback will be called with C<$fatal> true.

When on_connect is supplied, connect error are not fatal and AnyEvent::DBI
will not C<die>. You still cannot, however, use the $dbh object you
received from C<new> to make requests.

=item exec_server => 1

If you supply an C<exec_server> argument, then the DBI server process will
fork and exec another perl interpreter (using C<$^X>) with just the
AnyEvent::DBI proxy running. This will provide the cleanest possible proxy
for your database server.

If you do not supply the C<exec_server> argument (or supply it with a
false value) then the traditional method of starting the server by forking
the current process is used. The forked interpreter will try to clean
itself up by calling POSIX::close on all file descriptors except STDIN,
STDOUT, and STDERR (and the socket it uses to communicate with the cilent,
of course).

=item timeout => seconds

If you supply a timeout parameter (fractional values are supported), then
a timer is started any time the DBI handle expects a response from the
server. This includes connection setup as well as requests made to the
backend. The timeout spans the duration from the moment the first data
is written (or queued to be written) until all expected responses are
returned, but is postponed for "timeout" seconds each time more data is
returned from the server. If the timer ever goes off then a fatal error is
generated. If you have an C<on_error> handler installed, then it will be
called, otherwise your program will die().

When altering your databases with timeouts it is wise to use
transactions. If you quit due to timeout while performing insert, update
or schema-altering commands you can end up not knowing if the action was
submitted to the database, complicating recovery.

Timeout errors are always fatal.

=back

Any additional key-value pairs will be rolled into a hash reference
and passed as the final argument to the C<< DBI->connect (...) >>
call. For example, to supress errors on STDERR and send them instead to an
AnyEvent::Handle you could do:

   $dbh = new AnyEvent::DBI
              "DBI:mysql:test;mysql_read_default_file=/root/.my.cnf", "", "",
              PrintError => 0,
              on_error   => sub {
                 $log_handle->push_write ("DBI Error: $@ at $_[1]:$_[2]\n");
              };

=cut

# stupid Storable autoloading, total loss-loss situation
Storable::thaw Storable::freeze [];

sub new {
   my ($class, $dbi, $user, $pass, %arg) = @_;

   my ($client, $server) = AnyEvent::Util::portable_socketpair
      or croak "unable to create Anyevent::DBI communications pipe: $!";

   my %dbi_args = %arg;
   delete @dbi_args{qw(on_connect on_error timeout exec_server)};

   my $self = bless \%arg, $class;
   $self->{fh} = $client;

   AnyEvent::Util::fh_nonblocking $client, 1;

   my $rbuf;
   my @caller = (caller)[1,2]; # the "default" caller

   {
      Scalar::Util::weaken (my $self = $self);

      $self->{rw} = AnyEvent->io (fh => $client, poll => "r", cb => sub {
         return unless $self;

         $self->{last_activity} = AnyEvent->now;

         my $len = sysread $client, $rbuf, 65536, length $rbuf;

         if ($len > 0) {
            # we received data, so reset the timer

            while () {
               my $len = unpack "L", $rbuf;

               # full response available?
               last unless $len && $len + 4 <= length $rbuf;

               my $res = Storable::thaw substr $rbuf, 4;
               substr $rbuf, 0, $len + 4, ""; # remove length + request

               last unless $self;
               my $req = shift @{ $self->{queue} };

               if (defined $res->[0]) {
                  $res->[0] = $self;
                  $req->[0](@$res);
               } else {
                  my $cb = shift @$req;
                  local $@ = $res->[1];
                  $cb->($self);
                  $self->_error ($res->[1], @$req, $res->[2]) # error, request record, is_fatal
                     if $self; # cb() could have deleted it
               }

               # no more queued requests, so become idle
               undef $self->{last_activity}
                  if $self && !@{ $self->{queue} };
            }

         } elsif (defined $len) {
            # todo, caller?
            $self->_error ("unexpected eof", @caller, 1);
         } elsif ($! != Errno::EAGAIN) {
            # todo, caller?
            $self->_error ("read error: $!", @caller, 1);
         }
      });

      $self->{tw_cb} = sub {
         if ($self->{timeout} && $self->{last_activity}) {
            if (AnyEvent->now > $self->{last_activity} + $self->{timeout}) {
               # we did time out
               my $req = $self->{queue}[0];
               $self->_error (timeout => $req->[1], $req->[2], 1); # timeouts are always fatal
            } else {
               # we need to re-set the timeout watcher
               $self->{tw} = AnyEvent->timer (
                  after => $self->{last_activity} + $self->{timeout} - AnyEvent->now,
                  cb    => $self->{tw_cb},
               );
               Scalar::Util::weaken $self;
            }
         } else {
            # no timeout check wanted, or idle
            undef $self->{tw};
         }
      };

      $self->{ww_cb} = sub {
         return unless $self;

         $self->{last_activity} = AnyEvent->now;

         my $len = syswrite $client, $self->{wbuf}
            or return delete $self->{ww};

         substr $self->{wbuf}, 0, $len, "";
      };
   }

   my $pid = fork;

   if ($pid) {
      # parent
      close $server;
   } elsif (defined $pid) {
      # child
      my $serv_fno = fileno $server;

      if ($self->{exec_server}) {
         fcntl $server, &Fcntl::F_SETFD, 0; # don't close the server side
         exec {$^X}
              "$0 dbi slave",
              -e => "require shift; AnyEvent::DBI::serve_fd ($serv_fno, $VERSION)",
              $INC{"AnyEvent/DBI.pm"};
         POSIX::_exit 124;
      } else {
         ($_ != $serv_fno) && POSIX::close $_
            for $^F+1..$FD_MAX;
         serve_fh $server, $VERSION;

         # no other way on the broken windows platform, even this leaks
         # memory and might fail.
         kill 9, $$
            if AnyEvent::WIN32;

         # and this kills the parent process on windows
         POSIX::_exit 0;
      }
   } else {
      croak "fork: $!";
   }

   $self->{child_pid} = $pid;

   $self->_req (
      ($self->{on_connect} ? $self->{on_connect} : sub { }),
      (caller)[1,2],
      req_open => $dbi, $user, $pass, %dbi_args
   );

   $self
}

sub _server_pid {
   shift->{child_pid}
}

sub kill_child {
   my $self      = shift;
   my $child_pid = delete $self->{child_pid};
   if ($child_pid) {
      # send SIGKILL in two seconds
      my $murder_timer = AnyEvent->timer (
         after => 2,
         cb    => sub {
            kill 9, $child_pid;
         },
      );

      # reap process
      my $kid_watcher; $kid_watcher = AnyEvent->child (
         pid => $child_pid,
         cb  => sub {
            # just hold on to this so it won't go away
            undef $kid_watcher;
            # cancel SIGKILL
            undef $murder_timer;
         },
      );

      close $self->{fh};
   }
}

sub DESTROY {
   shift->kill_child;
}

sub _error {
   my ($self, $error, $filename, $line, $fatal) = @_;

   if ($fatal) {
      delete $self->{tw};
      delete $self->{rw};
      delete $self->{ww};
      delete $self->{fh};

      # for fatal errors call all enqueued callbacks with error
      while (my $req = shift @{$self->{queue}}) {
         local $@ = $error;
         $req->[0]->($self);
      }
      $self->kill_child;
   }

   local $@ = $error;

   if ($self->{on_error}) {
      $self->{on_error}($self, $filename, $line, $fatal)
   } else {
      die "$error at $filename, line $line\n";
   }
}

=item $dbh->on_error ($cb->($dbh, $filename, $line, $fatal))

Sets (or clears, with C<undef>) the C<on_error> handler.

=cut

sub on_error {
   $_[0]{on_error} = $_[1];
}

=item $dbh->timeout ($seconds)

Sets (or clears, with C<undef>) the database timeout. Useful to extend the
timeout when you are about to make a really long query.

=cut

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

   $self->{timeout} = $timeout;

   # reschedule timer if one was running
   $self->{tw_cb}->();
}

sub _req {
   my ($self, $cb, $filename, $line) = splice @_, 0, 4, ();

   unless ($self->{fh}) {
      local $@ = my $err = 'no database connection';
      $cb->($self);
      $self->_error ($err, $filename, $line, 1);
      return;
   }

   push @{ $self->{queue} }, [$cb, $filename, $line];

   # re-start timeout if necessary
   if ($self->{timeout} && !$self->{tw}) {
      $self->{last_activity} = AnyEvent->now;
      $self->{tw_cb}->();
   }

   $self->{wbuf} .= pack "L/a*", Storable::freeze \@_;

   unless ($self->{ww}) {
      my $len = syswrite $self->{fh}, $self->{wbuf};
      substr $self->{wbuf}, 0, $len, "";

      # still any left? then install a write watcher
      $self->{ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $self->{ww_cb})
         if length $self->{wbuf};
   }
}

=item $dbh->exec ("statement", @args, $cb->($dbh, \@rows, $rv))

Executes the given SQL statement with placeholders replaced by
C<@args>. The statement will be prepared and cached on the server side, so
using placeholders is extremely important.

The callback will be called with a weakened AnyEvent::DBI object as the
first argument and the result of C<fetchall_arrayref> as (or C<undef>
if the statement wasn't a select statement) as the second argument.

Third argument is the return value from the C<< DBI->execute >> method
call.

If an error occurs and the C<on_error> callback returns, then only C<$dbh>
will be passed and C<$@> contains the error message.

=item $dbh->attr ($attr_name[, $attr_value], $cb->($dbh, $new_value))

An accessor for the handle attributes, such as C<AutoCommit>,
C<RaiseError>, C<PrintError> and so on. If you provide an C<$attr_value>
(which might be C<undef>), then the given attribute will be set to that
value.

The callback will be passed the database handle and the attribute's value
if successful.

If an error occurs and the C<on_error> callback returns, then only C<$dbh>
will be passed and C<$@> contains the error message.

=item $dbh->begin_work ($cb->($dbh[, $rc]))

=item $dbh->commit     ($cb->($dbh[, $rc]))

=item $dbh->rollback   ($cb->($dbh[, $rc]))

The begin_work, commit, and rollback methods expose the equivalent
transaction control method of the DBI driver. On success, C<$rc> is true.

If an error occurs and the C<on_error> callback returns, then only C<$dbh>
will be passed and C<$@> contains the error message.

=item $dbh->func ('string_which_yields_args_when_evaled', $func_name, $cb->($dbh, $rc, $dbi_err, $dbi_errstr))

This gives access to database driver private methods. Because they
are not standard you cannot always depend on the value of C<$rc> or
C<$dbi_err>. Check the documentation for your specific driver/function
combination to see what it returns.

Note that the first argument will be eval'ed to produce the argument list to
the func() method. This must be done because the serialization protocol
between the AnyEvent::DBI server process and your program does not support the
passage of closures.

Here's an example to extend the query language in SQLite so it supports an
intstr() function:

    $cv = AnyEvent->condvar;
    $dbh->func (
       q{
          instr => 2, sub {
             my ($string, $search) = @_;
             return index $string, $search;
          },
       },
       create_function => sub {
          return $cv->send ($@)
             unless $#_;
          $cv->send (undef, @_[1,2,3]);
       }
    );

    my ($err,$rc,$errcode,$errstr) = $cv->recv;

    die $err if defined $err;
    die "EVAL failed: $errstr"
       if $errcode;

    # otherwise, we can ignore $rc and $errcode for this particular func

=cut

for my $cmd_name (qw(exec attr begin_work commit rollback func)) {
   eval 'sub ' . $cmd_name . '{
      my $cb = pop;
      splice @_, 1, 0, $cb, (caller)[1,2], "req_' . $cmd_name . '";
      &_req
   }';
}

=back

=head1 SEE ALSO

L<AnyEvent>, L<DBI>, L<Coro::Mysql>.

=head1 AUTHOR

   Marc Lehmann <schmorp@schmorp.de>
   http://home.schmorp.de/

   Adam Rosenstein <adam@redcondor.com>
   http://www.redcondor.com/

=cut

1;

Revision:	1.13
Committed:	Sat Oct 23 21:47:13 2010 UTC (13 years, 7 months ago) by root
Branch:	MAIN
Changes since 1.12:	+1 -1 lines
Log Message:	* empty log message *
#	User	Rev	Content
1	root	1.1	=head1 NAME
2
3			AnyEvent::DBI - asynchronous DBI access
4
5			=head1 SYNOPSIS
6
7			use AnyEvent::DBI;
8
9	root	1.5	my $cv = AnyEvent->condvar;
10
11			my $dbh = new AnyEvent::DBI "DBI:SQLite:dbname=test.db", "", "";
12
13			$dbh->exec ("select * from test where num=?", 10, sub {
14	root	1.11	my ($dbh, $rows, $rv) = @_;
15
16	root	1.12	$#_ or die "failure: $@";
17	root	1.5
18			print "@$_\n"
19			for @$rows;
20
21			$cv->broadcast;
22			});
23
24			# asynchronously do sth. else here
25
26			$cv->wait;
27
28	root	1.1	=head1 DESCRIPTION
29
30			This module is an L<AnyEvent> user, you need to make sure that you use and
31			run a supported event loop.
32
33	root	1.8	This module implements asynchronous DBI access by forking or executing
34	root	1.1	separate "DBI-Server" processes and sending them requests.
35
36			It means that you can run DBI requests in parallel to other tasks.
37
38	root	1.3	The overhead for very simple statements ("select 0") is somewhere
39	root	1.6	around 120% to 200% (dual/single core CPU) compared to an explicit
40	root	1.3	prepare_cached/execute/fetchrow_arrayref/finish combination.
41
42	root	1.11	=head2 ERROR HANDLING
43
44			This module defines a number of functions that accept a callback
45			argument. All callbacks used by this module get their AnyEvent::DBI handle
46			object passed as first argument.
47
48			If the request was successful, then there will be more arguments,
49			otherwise there will only be the C<$dbh> argument and C<$@> contains an
50			error message.
51
52			A convinient way to check whether an error occured is to check C<$#_> -
53			if that is true, then the function was successful, otherwise there was an
54			error.
55
56	root	1.1	=cut
57
58			package AnyEvent::DBI;
59
60	root	1.11	use strict qw(vars subs);
61	root	1.1	no warnings;
62
63			use Carp;
64			use Socket ();
65			use Scalar::Util ();
66			use Storable ();
67
68			use DBI ();
69
70			use AnyEvent ();
71			use AnyEvent::Util ();
72
73	root	1.11	use Errno ();
74			use Fcntl ();
75			use POSIX ();
76
77	root	1.12	our $VERSION = '2.0';
78	root	1.11
79			our $FD_MAX = eval { POSIX::sysconf (&POSIX::_SC_OPEN_MAX) - 1 } \|\| 1023;
80	root	1.1
81	root	1.11	# this is the forked server code, could/should be bundled as it's own file
82	root	1.1
83			our $DBH;
84
85			sub req_open {
86			my (undef, $dbi, $user, $pass, %attr) = @{+shift};
87
88	root	1.10	$DBH = DBI->connect ($dbi, $user, $pass, \%attr) or die $DBI::errstr;
89	root	1.1
90	root	1.11	[1, 1]
91	root	1.1	}
92
93	root	1.2	sub req_exec {
94			my (undef, $st, @args) = @{+shift};
95	root	1.10	my $sth = $DBH->prepare_cached ($st, undef, 1)
96			or die [$DBI::errstr];
97	root	1.2
98	root	1.7	my $rv = $sth->execute (@args)
99	root	1.10	or die [$sth->errstr];
100	root	1.2
101	root	1.11	[1, $sth->{NUM_OF_FIELDS} ? $sth->fetchall_arrayref : undef, $rv]
102	root	1.2	}
103
104	root	1.10	sub req_attr {
105	root	1.11	my (undef, $attr_name, @attr_val) = @{+shift};
106	root	1.10
107	root	1.11	$DBH->{$attr_name} = $attr_val[0]
108			if @attr_val;
109	root	1.10
110			[1, $DBH->{$attr_name}]
111			}
112
113			sub req_begin_work {
114	root	1.11	[1, $DBH->begin_work or die [$DBI::errstr]]
115	root	1.10	}
116
117			sub req_commit {
118	root	1.11	[1, $DBH->commit or die [$DBI::errstr]]
119	root	1.10	}
120
121			sub req_rollback {
122	root	1.11	[1, $DBH->rollback or die [$DBI::errstr]]
123	root	1.10	}
124
125			sub req_func {
126			my (undef, $arg_string, $function) = @{+shift};
127			my @args = eval $arg_string;
128
129	root	1.12	die "error evaling \$dbh->func() arg_string: $@"
130	root	1.11	if $@;
131	root	1.10
132	root	1.12	my $rc = $DBH->func (@args, $function);
133			return [1, $rc, $DBI::err, $DBI::errstr];
134	root	1.10	}
135
136	root	1.11	sub serve_fh($$) {
137			my ($fh, $version) = @_;
138	root	1.10
139	root	1.11	if ($VERSION != $version) {
140			syswrite $fh,
141			pack "L/a*",
142			Storable::freeze
143			[undef, "AnyEvent::DBI version mismatch ($VERSION vs. $version)"];
144			return;
145			}
146	root	1.1
147			eval {
148			my $rbuf;
149
150			while () {
151			sysread $fh, $rbuf, 16384, length $rbuf
152			or last;
153
154			while () {
155			my $len = unpack "L", $rbuf;
156
157			# full request available?
158			last unless $len && $len + 4 <= length $rbuf;
159
160			my $req = Storable::thaw substr $rbuf, 4;
161			substr $rbuf, 0, $len + 4, ""; # remove length + request
162
163			my $wbuf = eval { pack "L/a*", Storable::freeze $req->[0]($req) };
164	root	1.11	$wbuf = pack "L/a*", Storable::freeze [undef, ref $@ ? ("$@->[0]", $@->[1]) : ("$@", 1)]
165	root	1.1	if $@;
166
167			for (my $ofs = 0; $ofs < length $wbuf; ) {
168			$ofs += (syswrite $fh, substr $wbuf, $ofs
169			or die "unable to write results");
170			}
171			}
172			}
173			};
174	root	1.11	}
175	root	1.1
176	root	1.11	sub serve_fd($$) {
177			open my $fh, ">>&=$_[0]"
178			or die "Couldn't open server file descriptor: $!";
179	root	1.1
180	root	1.11	serve_fh $fh, $_[1];
181	root	1.10	}
182
183	root	1.1	=head2 METHODS
184
185			=over 4
186
187			=item $dbh = new AnyEvent::DBI $database, $user, $pass, [key => value]...
188
189			Returns a database handle for the given database. Each database handle
190			has an associated server process that executes statements in order. If
191			you want to run more than one statement in parallel, you need to create
192			additional database handles.
193
194			The advantage of this approach is that transactions work as state is
195			preserved.
196
197			Example:
198
199			$dbh = new AnyEvent::DBI
200			"DBI:mysql:test;mysql_read_default_file=/root/.my.cnf", "", "";
201
202			Additional key-value pairs can be used to adjust behaviour:
203
204			=over 4
205
206			=item on_error => $callback->($dbh, $filename, $line, $fatal)
207
208			When an error occurs, then this callback will be invoked. On entry, C<$@>
209			is set to the error message. C<$filename> and C<$line> is where the
210			original request was submitted.
211
212	root	1.11	If the fatal argument is true then the database connection is shut down
213			and your database handle became invalid. In addition to invoking the
214			C<on_error> callback, all of your queued request callbacks are called
215			without only the C<$dbh> argument.
216	root	1.1
217	root	1.2	If omitted, then C<die> will be called on any errors, fatal or not.
218	root	1.1
219	root	1.11	=item on_connect => $callback->($dbh[, $success])
220	root	1.10
221	root	1.11	If you supply an C<on_connect> callback, then this callback will be
222			invoked after the database connect attempt. If the connection succeeds,
223			C<$success> is true, otherwise it is missing and C<$@> contains the
224			C<$DBI::errstr>.
225
226			Regardless of whether C<on_connect> is supplied, connect errors will result in
227			C<on_error> being called. However, if no C<on_connect> callback is supplied, then
228			connection errors are considered fatal. The client will C<die> and the C<on_error>
229			callback will be called with C<$fatal> true.
230
231			When on_connect is supplied, connect error are not fatal and AnyEvent::DBI
232			will not C<die>. You still cannot, however, use the $dbh object you
233			received from C<new> to make requests.
234	root	1.10
235			=item exec_server => 1
236
237	root	1.11	If you supply an C<exec_server> argument, then the DBI server process will
238			fork and exec another perl interpreter (using C<$^X>) with just the
239	root	1.13	AnyEvent::DBI proxy running. This will provide the cleanest possible proxy
240	root	1.11	for your database server.
241
242			If you do not supply the C<exec_server> argument (or supply it with a
243			false value) then the traditional method of starting the server by forking
244			the current process is used. The forked interpreter will try to clean
245			itself up by calling POSIX::close on all file descriptors except STDIN,
246			STDOUT, and STDERR (and the socket it uses to communicate with the cilent,
247			of course).
248	root	1.10
249			=item timeout => seconds
250
251	root	1.11	If you supply a timeout parameter (fractional values are supported), then
252			a timer is started any time the DBI handle expects a response from the
253			server. This includes connection setup as well as requests made to the
254			backend. The timeout spans the duration from the moment the first data
255			is written (or queued to be written) until all expected responses are
256			returned, but is postponed for "timeout" seconds each time more data is
257			returned from the server. If the timer ever goes off then a fatal error is
258			generated. If you have an C<on_error> handler installed, then it will be
259			called, otherwise your program will die().
260
261			When altering your databases with timeouts it is wise to use
262			transactions. If you quit due to timeout while performing insert, update
263			or schema-altering commands you can end up not knowing if the action was
264			submitted to the database, complicating recovery.
265	root	1.10
266			Timeout errors are always fatal.
267
268	root	1.1	=back
269
270	root	1.11	Any additional key-value pairs will be rolled into a hash reference
271			and passed as the final argument to the C<< DBI->connect (...) >>
272			call. For example, to supress errors on STDERR and send them instead to an
273			AnyEvent::Handle you could do:
274
275			$dbh = new AnyEvent::DBI
276			"DBI:mysql:test;mysql_read_default_file=/root/.my.cnf", "", "",
277			PrintError => 0,
278			on_error => sub {
279			$log_handle->push_write ("DBI Error: $@ at $_[1]:$_[2]\n");
280			};
281	root	1.10
282	root	1.1	=cut
283
284			# stupid Storable autoloading, total loss-loss situation
285			Storable::thaw Storable::freeze [];
286
287			sub new {
288			my ($class, $dbi, $user, $pass, %arg) = @_;
289
290	root	1.11	my ($client, $server) = AnyEvent::Util::portable_socketpair
291			or croak "unable to create Anyevent::DBI communications pipe: $!";
292	root	1.1
293	root	1.11	my %dbi_args = %arg;
294			delete @dbi_args{qw(on_connect on_error timeout exec_server)};
295	root	1.10
296	root	1.1	my $self = bless \%arg, $class;
297			$self->{fh} = $client;
298
299			AnyEvent::Util::fh_nonblocking $client, 1;
300
301			my $rbuf;
302			my @caller = (caller)[1,2]; # the "default" caller
303
304	root	1.11	{
305			Scalar::Util::weaken (my $self = $self);
306
307			$self->{rw} = AnyEvent->io (fh => $client, poll => "r", cb => sub {
308			return unless $self;
309
310			$self->{last_activity} = AnyEvent->now;
311
312			my $len = sysread $client, $rbuf, 65536, length $rbuf;
313
314			if ($len > 0) {
315			# we received data, so reset the timer
316	root	1.1
317	root	1.11	while () {
318			my $len = unpack "L", $rbuf;
319	root	1.1
320	root	1.11	# full response available?
321			last unless $len && $len + 4 <= length $rbuf;
322	root	1.1
323	root	1.11	my $res = Storable::thaw substr $rbuf, 4;
324			substr $rbuf, 0, $len + 4, ""; # remove length + request
325	root	1.1
326	root	1.11	last unless $self;
327			my $req = shift @{ $self->{queue} };
328	root	1.1
329	root	1.11	if (defined $res->[0]) {
330			$res->[0] = $self;
331			$req->[0](@$res);
332			} else {
333			my $cb = shift @$req;
334			local $@ = $res->[1];
335			$cb->($self);
336			$self->_error ($res->[1], @$req, $res->[2]) # error, request record, is_fatal
337			if $self; # cb() could have deleted it
338	root	1.10	}
339	root	1.11
340			# no more queued requests, so become idle
341			undef $self->{last_activity}
342			if $self && !@{ $self->{queue} };
343	root	1.10	}
344
345	root	1.11	} elsif (defined $len) {
346			# todo, caller?
347			$self->_error ("unexpected eof", @caller, 1);
348			} elsif ($! != Errno::EAGAIN) {
349			# todo, caller?
350			$self->_error ("read error: $!", @caller, 1);
351			}
352			});
353
354			$self->{tw_cb} = sub {
355			if ($self->{timeout} && $self->{last_activity}) {
356			if (AnyEvent->now > $self->{last_activity} + $self->{timeout}) {
357			# we did time out
358			my $req = $self->{queue}[0];
359			$self->_error (timeout => $req->[1], $req->[2], 1); # timeouts are always fatal
360			} else {
361			# we need to re-set the timeout watcher
362			$self->{tw} = AnyEvent->timer (
363			after => $self->{last_activity} + $self->{timeout} - AnyEvent->now,
364			cb => $self->{tw_cb},
365			);
366			Scalar::Util::weaken $self;
367	root	1.1	}
368	root	1.11	} else {
369			# no timeout check wanted, or idle
370			undef $self->{tw};
371	root	1.1	}
372	root	1.11	};
373	root	1.1
374	root	1.11	$self->{ww_cb} = sub {
375			return unless $self;
376
377			$self->{last_activity} = AnyEvent->now;
378	root	1.1
379	root	1.11	my $len = syswrite $client, $self->{wbuf}
380			or return delete $self->{ww};
381	root	1.3
382	root	1.11	substr $self->{wbuf}, 0, $len, "";
383			};
384			}
385	root	1.3
386	root	1.1	my $pid = fork;
387
388			if ($pid) {
389			# parent
390			close $server;
391			} elsif (defined $pid) {
392			# child
393	root	1.10	my $serv_fno = fileno $server;
394	root	1.1
395	root	1.10	if ($self->{exec_server}) {
396	root	1.11	fcntl $server, &Fcntl::F_SETFD, 0; # don't close the server side
397			exec {$^X}
398			"$0 dbi slave",
399			-e => "require shift; AnyEvent::DBI::serve_fd ($serv_fno, $VERSION)",
400			$INC{"AnyEvent/DBI.pm"};
401	root	1.10	POSIX::_exit 124;
402			} else {
403			($_ != $serv_fno) && POSIX::close $_
404	root	1.11	for $^F+1..$FD_MAX;
405			serve_fh $server, $VERSION;
406
407			# no other way on the broken windows platform, even this leaks
408			# memory and might fail.
409			kill 9, $$
410			if AnyEvent::WIN32;
411
412			# and this kills the parent process on windows
413			POSIX::_exit 0;
414	root	1.10	}
415	root	1.1	} else {
416			croak "fork: $!";
417			}
418
419	root	1.10	$self->{child_pid} = $pid;
420	root	1.11
421	root	1.10	$self->_req (
422			($self->{on_connect} ? $self->{on_connect} : sub { }),
423			(caller)[1,2],
424			req_open => $dbi, $user, $pass, %dbi_args
425			);
426	root	1.1
427			$self
428			}
429
430	root	1.10	sub _server_pid {
431			shift->{child_pid}
432			}
433
434			sub kill_child {
435			my $self = shift;
436			my $child_pid = delete $self->{child_pid};
437			if ($child_pid) {
438			# send SIGKILL in two seconds
439			my $murder_timer = AnyEvent->timer (
440			after => 2,
441			cb => sub {
442			kill 9, $child_pid;
443			},
444			);
445
446			# reap process
447	root	1.11	my $kid_watcher; $kid_watcher = AnyEvent->child (
448			pid => $child_pid,
449	root	1.10	cb => sub {
450			# just hold on to this so it won't go away
451			undef $kid_watcher;
452			# cancel SIGKILL
453			undef $murder_timer;
454			},
455			);
456
457	root	1.11	close $self->{fh};
458	root	1.10	}
459			}
460
461			sub DESTROY {
462			shift->kill_child;
463			}
464
465	root	1.1	sub _error {
466			my ($self, $error, $filename, $line, $fatal) = @_;
467
468	root	1.10	if ($fatal) {
469	root	1.11	delete $self->{tw};
470	root	1.10	delete $self->{rw};
471			delete $self->{ww};
472			delete $self->{fh};
473
474			# for fatal errors call all enqueued callbacks with error
475			while (my $req = shift @{$self->{queue}}) {
476	root	1.11	local $@ = $error;
477			$req->[0]->($self);
478	root	1.10	}
479			$self->kill_child;
480			}
481	root	1.1
482	root	1.11	local $@ = $error;
483	root	1.1
484	root	1.9	if ($self->{on_error}) {
485	root	1.10	$self->{on_error}($self, $filename, $line, $fatal)
486			} else {
487			die "$error at $filename, line $line\n";
488	root	1.9	}
489	root	1.10	}
490
491			=item $dbh->on_error ($cb->($dbh, $filename, $line, $fatal))
492
493	root	1.11	Sets (or clears, with C<undef>) the C<on_error> handler.
494	root	1.10
495			=cut
496	root	1.1
497	root	1.10	sub on_error {
498			$_[0]{on_error} = $_[1];
499			}
500
501			=item $dbh->timeout ($seconds)
502
503			Sets (or clears, with C<undef>) the database timeout. Useful to extend the
504			timeout when you are about to make a really long query.
505
506			=cut
507
508			sub timeout {
509			my ($self, $timeout) = @_;
510
511	root	1.11	$self->{timeout} = $timeout;
512	root	1.10
513	root	1.11	# reschedule timer if one was running
514			$self->{tw_cb}->();
515	root	1.1	}
516
517			sub _req {
518	root	1.10	my ($self, $cb, $filename, $line) = splice @_, 0, 4, ();
519
520	root	1.11	unless ($self->{fh}) {
521			local $@ = my $err = 'no database connection';
522			$cb->($self);
523	root	1.10	$self->_error ($err, $filename, $line, 1);
524			return;
525			}
526
527	root	1.11	push @{ $self->{queue} }, [$cb, $filename, $line];
528	root	1.1
529	root	1.11	# re-start timeout if necessary
530			if ($self->{timeout} && !$self->{tw}) {
531			$self->{last_activity} = AnyEvent->now;
532			$self->{tw_cb}->();
533	root	1.10	}
534	root	1.1
535			$self->{wbuf} .= pack "L/a*", Storable::freeze \@_;
536
537			unless ($self->{ww}) {
538			my $len = syswrite $self->{fh}, $self->{wbuf};
539			substr $self->{wbuf}, 0, $len, "";
540
541			# still any left? then install a write watcher
542			$self->{ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $self->{ww_cb})
543			if length $self->{wbuf};
544			}
545			}
546
547	root	1.11	=item $dbh->exec ("statement", @args, $cb->($dbh, \@rows, $rv))
548	root	1.1
549			Executes the given SQL statement with placeholders replaced by
550	root	1.2	C<@args>. The statement will be prepared and cached on the server side, so
551	root	1.11	using placeholders is extremely important.
552	root	1.1
553	root	1.11	The callback will be called with a weakened AnyEvent::DBI object as the
554			first argument and the result of C<fetchall_arrayref> as (or C<undef>
555			if the statement wasn't a select statement) as the second argument.
556	root	1.1
557	root	1.11	Third argument is the return value from the C<< DBI->execute >> method
558			call.
559
560			If an error occurs and the C<on_error> callback returns, then only C<$dbh>
561	root	1.2	will be passed and C<$@> contains the error message.
562
563	root	1.11	=item $dbh->attr ($attr_name[, $attr_value], $cb->($dbh, $new_value))
564
565			An accessor for the handle attributes, such as C<AutoCommit>,
566			C<RaiseError>, C<PrintError> and so on. If you provide an C<$attr_value>
567			(which might be C<undef>), then the given attribute will be set to that
568			value.
569	root	1.10
570	root	1.11	The callback will be passed the database handle and the attribute's value
571			if successful.
572
573			If an error occurs and the C<on_error> callback returns, then only C<$dbh>
574			will be passed and C<$@> contains the error message.
575	root	1.10
576	root	1.12	=item $dbh->begin_work ($cb->($dbh[, $rc]))
577	root	1.10
578	root	1.12	=item $dbh->commit ($cb->($dbh[, $rc]))
579	root	1.10
580	root	1.12	=item $dbh->rollback ($cb->($dbh[, $rc]))
581	root	1.10
582	root	1.11	The begin_work, commit, and rollback methods expose the equivalent
583	root	1.12	transaction control method of the DBI driver. On success, C<$rc> is true.
584	root	1.10
585	root	1.11	If an error occurs and the C<on_error> callback returns, then only C<$dbh>
586			will be passed and C<$@> contains the error message.
587	root	1.10
588	root	1.12	=item $dbh->func ('string_which_yields_args_when_evaled', $func_name, $cb->($dbh, $rc, $dbi_err, $dbi_errstr))
589	root	1.10
590	root	1.11	This gives access to database driver private methods. Because they
591	root	1.12	are not standard you cannot always depend on the value of C<$rc> or
592			C<$dbi_err>. Check the documentation for your specific driver/function
593			combination to see what it returns.
594	root	1.10
595			Note that the first argument will be eval'ed to produce the argument list to
596	root	1.11	the func() method. This must be done because the serialization protocol
597	root	1.10	between the AnyEvent::DBI server process and your program does not support the
598			passage of closures.
599
600			Here's an example to extend the query language in SQLite so it supports an
601			intstr() function:
602
603			$cv = AnyEvent->condvar;
604	root	1.11	$dbh->func (
605	root	1.10	q{
606	root	1.11	instr => 2, sub {
607	root	1.10	my ($string, $search) = @_;
608			return index $string, $search;
609			},
610			},
611	root	1.11	create_function => sub {
612	root	1.12	return $cv->send ($@)
613			unless $#_;
614			$cv->send (undef, @_[1,2,3]);
615	root	1.11	}
616	root	1.10	);
617	root	1.11
618	root	1.12	my ($err,$rc,$errcode,$errstr) = $cv->recv;
619	root	1.11
620	root	1.12	die $err if defined $err;
621			die "EVAL failed: $errstr"
622			if $errcode;
623	root	1.11
624	root	1.12	# otherwise, we can ignore $rc and $errcode for this particular func
625	root	1.10
626	root	1.1	=cut
627
628	root	1.10	for my $cmd_name (qw(exec attr begin_work commit rollback func)) {
629			eval 'sub ' . $cmd_name . '{
630			my $cb = pop;
631			splice @_, 1, 0, $cb, (caller)[1,2], "req_' . $cmd_name . '";
632	root	1.11	&_req
633	root	1.10	}';
634	root	1.1	}
635
636			=back
637
638			=head1 SEE ALSO
639
640	root	1.11	L<AnyEvent>, L<DBI>, L<Coro::Mysql>.
641	root	1.1
642			=head1 AUTHOR
643
644	root	1.4	Marc Lehmann <schmorp@schmorp.de>
645			http://home.schmorp.de/
646	root	1.1
647	root	1.10	Adam Rosenstein <adam@redcondor.com>
648			http://www.redcondor.com/
649
650	root	1.1	=cut
651
652	root	1.10	1;
653	root	1.1