ViewVC Help
View File | Revision Log | Show Annotations | Download File
/cvs/AnyEvent-DBI/DBI.pm
(Generate patch)

Comparing AnyEvent-DBI/DBI.pm (file contents):
Revision 1.16 by root, Tue Apr 2 02:13:13 2013 UTC vs.
Revision 1.20 by root, Sat Mar 24 23:22:25 2018 UTC

33This module implements asynchronous DBI access by forking or executing 33This module implements asynchronous DBI access by forking or executing
34separate "DBI-Server" processes and sending them requests. 34separate "DBI-Server" processes and sending them requests.
35 35
36It means that you can run DBI requests in parallel to other tasks. 36It means that you can run DBI requests in parallel to other tasks.
37 37
38The overhead for very simple statements ("select 0") is somewhere 38With DBD::mysql, the overhead for very simple statements
39around 100% to 120% (dual/single core CPU) compared to an explicit 39("select 0") is somewhere around 50% compared to an explicit
40prepare_cached/execute/fetchrow_arrayref/finish combination. 40prepare_cached/execute/fetchrow_arrayref/finish combination. With
41DBD::SQlite3, it's more like a factor of 8 for this trivial statement.
41 42
42=head2 ERROR HANDLING 43=head2 ERROR HANDLING
43 44
44This module defines a number of functions that accept a callback 45This module defines a number of functions that accept a callback
45argument. All callbacks used by this module get their AnyEvent::DBI handle 46argument. All callbacks used by this module get their AnyEvent::DBI handle
47 48
48If the request was successful, then there will be more arguments, 49If the request was successful, then there will be more arguments,
49otherwise there will only be the C<$dbh> argument and C<$@> contains an 50otherwise there will only be the C<$dbh> argument and C<$@> contains an
50error message. 51error message.
51 52
52A convinient way to check whether an error occured is to check C<$#_> - 53A convenient way to check whether an error occurred is to check C<$#_> -
53if that is true, then the function was successful, otherwise there was an 54if that is true, then the function was successful, otherwise there was an
54error. 55error.
55 56
56=cut 57=cut
57 58
58package AnyEvent::DBI; 59package AnyEvent::DBI;
59 60
60use common::sense; 61use common::sense;
61 62
62use Carp; 63use Carp;
63use Socket (); 64use Convert::Scalar ();
64use Scalar::Util (); 65use AnyEvent::Fork ();
65use Storable (); 66use CBOR::XS ();
66
67use DBI (); # only needed in child actually - do it before fork & !exec?
68 67
69use AnyEvent (); 68use AnyEvent ();
70use AnyEvent::Util (); 69use AnyEvent::Util ();
71 70
72use Errno (); 71use Errno ();
73use Fcntl ();
74use POSIX ();
75 72
76our $VERSION = '2.3'; 73our $VERSION = '3.02';
77
78our $FD_MAX = eval { POSIX::sysconf (&POSIX::_SC_OPEN_MAX) - 1 } || 1023;
79
80# this is the forked server code, could/should be bundled as it's own file
81
82our $DBH;
83
84sub req_open {
85 my (undef, $dbi, $user, $pass, %attr) = @{+shift};
86
87 $DBH = DBI->connect ($dbi, $user, $pass, \%attr) or die $DBI::errstr;
88
89 [1, 1]
90}
91
92sub req_exec {
93 my (undef, $st, @args) = @{+shift};
94 my $sth = $DBH->prepare_cached ($st, undef, 1)
95 or die [$DBI::errstr];
96
97 my $rv = $sth->execute (@args)
98 or die [$sth->errstr];
99
100 [1, $sth->{NUM_OF_FIELDS} ? $sth->fetchall_arrayref : undef, $rv]
101}
102
103sub req_attr {
104 my (undef, $attr_name, @attr_val) = @{+shift};
105
106 $DBH->{$attr_name} = $attr_val[0]
107 if @attr_val;
108
109 [1, $DBH->{$attr_name}]
110}
111
112sub req_begin_work {
113 [1, $DBH->begin_work || die [$DBI::errstr]]
114}
115
116sub req_commit {
117 [1, $DBH->commit || die [$DBI::errstr]]
118}
119
120sub req_rollback {
121 [1, $DBH->rollback || die [$DBI::errstr]]
122}
123
124sub req_func {
125 my (undef, $arg_string, $function) = @{+shift};
126 my @args = eval $arg_string;
127
128 die "error evaling \$dbh->func() arg_string: $@"
129 if $@;
130
131 my $rc = $DBH->func (@args, $function);
132 return [1, $rc, $DBI::err, $DBI::errstr];
133}
134
135sub serve_fh($$) {
136 my ($fh, $version) = @_;
137
138 if ($VERSION != $version) {
139 syswrite $fh,
140 pack "L/a*",
141 Storable::freeze
142 [undef, "AnyEvent::DBI version mismatch ($VERSION vs. $version)"];
143 return;
144 }
145
146 eval {
147 my $rbuf;
148
149 while () {
150 sysread $fh, $rbuf, 16384, length $rbuf
151 or last;
152
153 while () {
154 my $len = unpack "L", $rbuf;
155
156 # full request available?
157 last unless $len && $len + 4 <= length $rbuf;
158
159 my $req = Storable::thaw substr $rbuf, 4;
160 substr $rbuf, 0, $len + 4, ""; # remove length + request
161
162 my $wbuf = eval { pack "L/a*", Storable::freeze $req->[0]($req) };
163 $wbuf = pack "L/a*", Storable::freeze [undef, ref $@ ? ("$@->[0]", $@->[1]) : ("$@", 1)]
164 if $@;
165
166 for (my $ofs = 0; $ofs < length $wbuf; ) {
167 $ofs += (syswrite $fh, substr $wbuf, $ofs
168 or die "unable to write results");
169 }
170 }
171 }
172 };
173}
174
175sub serve_fd($$) {
176 open my $fh, ">>&=$_[0]"
177 or die "Couldn't open server file descriptor: $!";
178
179 serve_fh $fh, $_[1];
180}
181 74
182=head2 METHODS 75=head2 METHODS
183 76
184=over 4 77=over 4
185 78
229 122
230When on_connect is supplied, connect error are not fatal and AnyEvent::DBI 123When on_connect is supplied, connect error are not fatal and AnyEvent::DBI
231will not C<die>. You still cannot, however, use the $dbh object you 124will not C<die>. You still cannot, however, use the $dbh object you
232received from C<new> to make requests. 125received from C<new> to make requests.
233 126
234=item exec_server => 1 127=item fork_template => $AnyEvent::Fork-object
235 128
236If you supply an C<exec_server> argument, then the DBI server process will 129C<AnyEvent::DBI> uses C<< AnyEvent::Fork->new >> to create the database
237fork and exec another perl interpreter (using C<$^X>) with just the 130slave, which in turn either C<exec>'s a new process (similar to the old
238AnyEvent::DBI proxy running. This will provide the cleanest possible proxy 131C<exec_server> constructor argument) or uses a process forked early (see
239for your database server. 132L<AnyEvent::Fork::Early>).
240 133
241If you do not supply the C<exec_server> argument (or supply it with a 134With this argument you can provide your own fork template. This can be
242false value) then the traditional method of starting the server by forking 135useful if you create a lot of C<AnyEvent::DBI> handles and want to save
243the current process is used. The forked interpreter will try to clean 136memory (And speed up startup) by not having to load C<AnyEvent::DBI> again
244itself up by calling POSIX::close on all file descriptors except STDIN, 137and again into your child processes:
245STDOUT, and STDERR (and the socket it uses to communicate with the cilent, 138
246of course). 139 my $template = AnyEvent::Fork
140 ->new # create new template
141 ->require ("AnyEvent::DBI::Slave"); # preload AnyEvent::DBI::Slave module
142
143 for (...) {
144 $dbh = new AnyEvent::DBI ...
145 fork_template => $template;
247 146
248=item timeout => seconds 147=item timeout => seconds
249 148
250If you supply a timeout parameter (fractional values are supported), then 149If you supply a timeout parameter (fractional values are supported), then
251a timer is started any time the DBI handle expects a response from the 150a timer is started any time the DBI handle expects a response from the
266 165
267=back 166=back
268 167
269Any additional key-value pairs will be rolled into a hash reference 168Any additional key-value pairs will be rolled into a hash reference
270and passed as the final argument to the C<< DBI->connect (...) >> 169and passed as the final argument to the C<< DBI->connect (...) >>
271call. For example, to supress errors on STDERR and send them instead to an 170call. For example, to suppress errors on STDERR and send them instead to an
272AnyEvent::Handle you could do: 171AnyEvent::Handle you could do:
273 172
274 $dbh = new AnyEvent::DBI 173 $dbh = new AnyEvent::DBI
275 "DBI:mysql:test;mysql_read_default_file=/root/.my.cnf", "", "", 174 "DBI:mysql:test;mysql_read_default_file=/root/.my.cnf", "", "",
276 PrintError => 0, 175 PrintError => 0,
278 $log_handle->push_write ("DBI Error: $@ at $_[1]:$_[2]\n"); 177 $log_handle->push_write ("DBI Error: $@ at $_[1]:$_[2]\n");
279 }; 178 };
280 179
281=cut 180=cut
282 181
283# stupid Storable autoloading, total loss-loss situation
284Storable::thaw Storable::freeze [];
285
286sub new { 182sub new {
287 my ($class, $dbi, $user, $pass, %arg) = @_; 183 my ($class, $dbi, $user, $pass, %arg) = @_;
288 184
185 # we use our own socketpair, so we always have a socket
186 # available, even before the forked process exsist.
187 # this is mostly done so this module is compatible
188 # to versions of itself older than 3.0.
289 my ($client, $server) = AnyEvent::Util::portable_socketpair 189 my ($client, $server) = AnyEvent::Util::portable_socketpair
290 or croak "unable to create AnyEvent::DBI communications pipe: $!"; 190 or croak "unable to create AnyEvent::DBI communications pipe: $!";
291 191
192 AnyEvent::fh_unblock $client;
193
194 my $fork = delete $arg{fork_template};
195
292 my %dbi_args = %arg; 196 my %dbi_args = %arg;
293 delete @dbi_args{qw(on_connect on_error timeout exec_server)}; 197 delete @dbi_args{qw(on_connect on_error timeout fork_template exec_server)};
294 198
295 my $self = bless \%arg, $class; 199 my $self = bless \%arg, $class;
200
296 $self->{fh} = $client; 201 $self->{fh} = $client;
297
298 AnyEvent::Util::fh_nonblocking $client, 1;
299 202
300 my $rbuf; 203 my $rbuf;
301 my @caller = (caller)[1,2]; # the "default" caller 204 my @caller = (caller)[1,2]; # the "default" caller
302 205
206 $fork = $fork ? $fork->fork : AnyEvent::Fork->new
207 or croak "fork: $!";
208
209 $fork->require ("AnyEvent::DBI::Slave");
210 $fork->send_arg ($VERSION);
211 $fork->send_fh ($server);
212
213 # we don't rely on the callback, because we use our own
214 # socketpair, for better or worse.
215 $fork->run ("AnyEvent::DBI::Slave::serve", sub { });
216
303 { 217 {
304 Scalar::Util::weaken (my $self = $self); 218 Convert::Scalar::weaken (my $self = $self);
219
220 my $cbor = new CBOR::XS;
305 221
306 $self->{rw} = AE::io $client, 0, sub { 222 $self->{rw} = AE::io $client, 0, sub {
307 return unless $self; 223 my $len = Convert::Scalar::extend_read $client, $rbuf, 65536;
308
309 my $len = sysread $client, $rbuf, 65536, length $rbuf;
310 224
311 if ($len > 0) { 225 if ($len > 0) {
312 # we received data, so reset the timer 226 # we received data, so reset the timer
313 $self->{last_activity} = AE::now; 227 $self->{last_activity} = AE::now;
314 228
315 while () { 229 for my $res ($cbor->incr_parse_multiple ($rbuf)) {
316 my $len = unpack "L", $rbuf;
317
318 # full response available?
319 last unless $len && $len + 4 <= length $rbuf;
320
321 my $res = Storable::thaw substr $rbuf, 4;
322 substr $rbuf, 0, $len + 4, ""; # remove length + request
323
324 last unless $self; 230 last unless $self;
231
325 my $req = shift @{ $self->{queue} }; 232 my $req = shift @{ $self->{queue} };
326 233
327 if (defined $res->[0]) { 234 if (defined $res->[0]) {
328 $res->[0] = $self; 235 $res->[0] = $self;
329 $req->[0](@$res); 236 $req->[0](@$res);
370 undef $self->{tw}; 277 undef $self->{tw};
371 } 278 }
372 }; 279 };
373 280
374 $self->{ww_cb} = sub { 281 $self->{ww_cb} = sub {
375 return unless $self;
376
377 $self->{last_activity} = AE::now; 282 $self->{last_activity} = AE::now;
378 283
379 my $len = syswrite $client, $self->{wbuf} 284 my $len = syswrite $client, $self->{wbuf}
380 or return delete $self->{ww}; 285 or return delete $self->{ww};
381 286
382 substr $self->{wbuf}, 0, $len, ""; 287 substr $self->{wbuf}, 0, $len, "";
383 }; 288 };
384 } 289 }
385 290
386 my $pid = fork;
387
388 if ($pid) {
389 # parent
390 close $server;
391 } elsif (defined $pid) {
392 # child
393 my $serv_fno = fileno $server;
394
395 if ($self->{exec_server}) {
396 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 POSIX::_exit 124;
402 } else {
403 ($_ != $serv_fno) && POSIX::close $_
404 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 }
415 } else {
416 croak "fork: $!";
417 }
418
419 $self->{child_pid} = $pid;
420
421 $self->_req ( 291 $self->_req (
292 sub {
293 return unless $self;
294 $self->{child_pid} = $_[1];
295 },
296 (caller)[1,2],
297 "req_pid"
298 );
299
300 $self->_req (
301 sub {
302 return unless $self;
422 ($self->{on_connect} ? $self->{on_connect} : sub { }), 303 &{ $self->{on_connect} } if $self->{on_connect};
304 },
423 (caller)[1,2], 305 (caller)[1,2],
424 req_open => $dbi, $user, $pass, %dbi_args 306 req_open => $dbi, $user, $pass, %dbi_args
425 ); 307 );
426 308
427 $self 309 $self
440 undef $kid_watcher; 322 undef $kid_watcher;
441 }; 323 };
442 kill TERM => $pid; 324 kill TERM => $pid;
443 } 325 }
444 326
327 delete $self->{rw};
328 delete $self->{ww};
329 delete $self->{tw};
445 close delete $self->{fh}; 330 close delete $self->{fh};
446} 331}
447 332
448sub DESTROY { 333sub DESTROY {
449 shift->kill_child; 334 shift->kill_child;
517 if ($self->{timeout} && !$self->{tw}) { 402 if ($self->{timeout} && !$self->{tw}) {
518 $self->{last_activity} = AE::now; 403 $self->{last_activity} = AE::now;
519 $self->{tw_cb}->(); 404 $self->{tw_cb}->();
520 } 405 }
521 406
522 $self->{wbuf} .= pack "L/a*", Storable::freeze \@_; 407 $self->{wbuf} .= CBOR::XS::encode_cbor \@_;
523 408
524 unless ($self->{ww}) { 409 unless ($self->{ww}) {
525 my $len = syswrite $self->{fh}, $self->{wbuf}; 410 my $len = syswrite $self->{fh}, $self->{wbuf};
526 substr $self->{wbuf}, 0, $len, ""; 411 substr $self->{wbuf}, 0, $len, "";
527 412
529 $self->{ww} = AE::io $self->{fh}, 1, $self->{ww_cb} 414 $self->{ww} = AE::io $self->{fh}, 1, $self->{ww_cb}
530 if length $self->{wbuf}; 415 if length $self->{wbuf};
531 } 416 }
532} 417}
533 418
419=item $dbh->attr ($attr_name[, $attr_value], $cb->($dbh, $new_value))
420
421An accessor for the database handle attributes, such as C<AutoCommit>,
422C<RaiseError>, C<PrintError> and so on. If you provide an C<$attr_value>
423(which might be C<undef>), then the given attribute will be set to that
424value.
425
426The callback will be passed the database handle and the attribute's value
427if successful.
428
429If an error occurs and the C<on_error> callback returns, then only C<$dbh>
430will be passed and C<$@> contains the error message.
431
534=item $dbh->exec ("statement", @args, $cb->($dbh, \@rows, $rv)) 432=item $dbh->exec ("statement", @args, $cb->($dbh, \@rows, $rv))
535 433
536Executes the given SQL statement with placeholders replaced by 434Executes the given SQL statement with placeholders replaced by
537C<@args>. The statement will be prepared and cached on the server side, so 435C<@args>. The statement will be prepared and cached on the server side, so
538using placeholders is extremely important. 436using placeholders is extremely important.
545call. 443call.
546 444
547If an error occurs and the C<on_error> callback returns, then only C<$dbh> 445If an error occurs and the C<on_error> callback returns, then only C<$dbh>
548will be passed and C<$@> contains the error message. 446will be passed and C<$@> contains the error message.
549 447
550=item $dbh->attr ($attr_name[, $attr_value], $cb->($dbh, $new_value)) 448=item $dbh->stattr ($attr_name, $cb->($dbh, $value))
551 449
552An accessor for the handle attributes, such as C<AutoCommit>, 450An accessor for the statement attributes of the most recently executed
553C<RaiseError>, C<PrintError> and so on. If you provide an C<$attr_value> 451statement, such as C<NAME> or C<TYPE>.
554(which might be C<undef>), then the given attribute will be set to that
555value.
556 452
557The callback will be passed the database handle and the attribute's value 453The callback will be passed the database handle and the attribute's value
558if successful. 454if successful.
559 455
560If an error occurs and the C<on_error> callback returns, then only C<$dbh> 456If an error occurs and the C<on_error> callback returns, then only C<$dbh>
610 506
611 # otherwise, we can ignore $rc and $errcode for this particular func 507 # otherwise, we can ignore $rc and $errcode for this particular func
612 508
613=cut 509=cut
614 510
615for my $cmd_name (qw(exec attr begin_work commit rollback func)) { 511for my $cmd_name (qw(attr exec stattr begin_work commit rollback func)) {
616 eval 'sub ' . $cmd_name . '{ 512 eval 'sub ' . $cmd_name . '{
617 my $cb = pop; 513 my $cb = pop;
618 splice @_, 1, 0, $cb, (caller)[1,2], "req_' . $cmd_name . '"; 514 splice @_, 1, 0, $cb, (caller)[1,2], "req_' . $cmd_name . '";
619 &_req 515 &_req
620 }'; 516 }';
624 520
625=head1 SEE ALSO 521=head1 SEE ALSO
626 522
627L<AnyEvent>, L<DBI>, L<Coro::Mysql>. 523L<AnyEvent>, L<DBI>, L<Coro::Mysql>.
628 524
629=head1 AUTHOR 525=head1 AUTHOR AND CONTACT
630 526
631 Marc Lehmann <schmorp@schmorp.de> 527 Marc Lehmann <schmorp@schmorp.de> (current maintainer)
632 http://home.schmorp.de/ 528 http://home.schmorp.de/
633 529
634 Adam Rosenstein <adam@redcondor.com> 530 Adam Rosenstein <adam@redcondor.com>
635 http://www.redcondor.com/ 531 http://www.redcondor.com/
636 532
637=cut 533=cut
638 534
6391; 5351
640

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines