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.17 by root, Tue Aug 15 07:28:55 2017 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.0';
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 my $fork = delete $arg{fork_template};
193
292 my %dbi_args = %arg; 194 my %dbi_args = %arg;
293 delete @dbi_args{qw(on_connect on_error timeout exec_server)}; 195 delete @dbi_args{qw(on_connect on_error timeout fork_template exec_server)};
294 196
295 my $self = bless \%arg, $class; 197 my $self = bless \%arg, $class;
198
296 $self->{fh} = $client; 199 $self->{fh} = $client;
297
298 AnyEvent::Util::fh_nonblocking $client, 1;
299 200
300 my $rbuf; 201 my $rbuf;
301 my @caller = (caller)[1,2]; # the "default" caller 202 my @caller = (caller)[1,2]; # the "default" caller
302 203
204 $fork = $fork ? $fork->fork : AnyEvent::Fork->new
205 or croak "fork: $!";
206
207 $fork->require ("AnyEvent::DBI::Slave");
208 $fork->send_arg ($VERSION);
209 $fork->send_fh ($server);
210
211 # we don't rely on the callback, because we use our own
212 # socketpair, for better or worse.
213 $fork->run ("AnyEvent::DBI::Slave::serve", sub { });
214
303 { 215 {
304 Scalar::Util::weaken (my $self = $self); 216 Convert::Scalar::weaken (my $self = $self);
217
218 my $cbor = new CBOR::XS;
305 219
306 $self->{rw} = AE::io $client, 0, sub { 220 $self->{rw} = AE::io $client, 0, sub {
307 return unless $self; 221 my $len = Convert::Scalar::extend_read $client, $rbuf, 65536;
308
309 my $len = sysread $client, $rbuf, 65536, length $rbuf;
310 222
311 if ($len > 0) { 223 if ($len > 0) {
312 # we received data, so reset the timer 224 # we received data, so reset the timer
313 $self->{last_activity} = AE::now; 225 $self->{last_activity} = AE::now;
314 226
315 while () { 227 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; 228 last unless $self;
229
325 my $req = shift @{ $self->{queue} }; 230 my $req = shift @{ $self->{queue} };
326 231
327 if (defined $res->[0]) { 232 if (defined $res->[0]) {
328 $res->[0] = $self; 233 $res->[0] = $self;
329 $req->[0](@$res); 234 $req->[0](@$res);
370 undef $self->{tw}; 275 undef $self->{tw};
371 } 276 }
372 }; 277 };
373 278
374 $self->{ww_cb} = sub { 279 $self->{ww_cb} = sub {
375 return unless $self;
376
377 $self->{last_activity} = AE::now; 280 $self->{last_activity} = AE::now;
378 281
379 my $len = syswrite $client, $self->{wbuf} 282 my $len = syswrite $client, $self->{wbuf}
380 or return delete $self->{ww}; 283 or return delete $self->{ww};
381 284
382 substr $self->{wbuf}, 0, $len, ""; 285 substr $self->{wbuf}, 0, $len, "";
383 }; 286 };
384 } 287 }
385 288
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 ( 289 $self->_req (
290 sub {
291 return unless $self;
292 $self->{child_pid} = $_[1];
293 },
294 (caller)[1,2],
295 "req_pid"
296 );
297
298 $self->_req (
299 sub {
300 return unless $self;
422 ($self->{on_connect} ? $self->{on_connect} : sub { }), 301 &{ $self->{on_connect} } if $self->{on_connect};
302 },
423 (caller)[1,2], 303 (caller)[1,2],
424 req_open => $dbi, $user, $pass, %dbi_args 304 req_open => $dbi, $user, $pass, %dbi_args
425 ); 305 );
426 306
427 $self 307 $self
440 undef $kid_watcher; 320 undef $kid_watcher;
441 }; 321 };
442 kill TERM => $pid; 322 kill TERM => $pid;
443 } 323 }
444 324
325 delete $self->{rw};
326 delete $self->{ww};
327 delete $self->{tw};
445 close delete $self->{fh}; 328 close delete $self->{fh};
446} 329}
447 330
448sub DESTROY { 331sub DESTROY {
449 shift->kill_child; 332 shift->kill_child;
517 if ($self->{timeout} && !$self->{tw}) { 400 if ($self->{timeout} && !$self->{tw}) {
518 $self->{last_activity} = AE::now; 401 $self->{last_activity} = AE::now;
519 $self->{tw_cb}->(); 402 $self->{tw_cb}->();
520 } 403 }
521 404
522 $self->{wbuf} .= pack "L/a*", Storable::freeze \@_; 405 $self->{wbuf} .= CBOR::XS::encode_cbor \@_;
523 406
524 unless ($self->{ww}) { 407 unless ($self->{ww}) {
525 my $len = syswrite $self->{fh}, $self->{wbuf}; 408 my $len = syswrite $self->{fh}, $self->{wbuf};
526 substr $self->{wbuf}, 0, $len, ""; 409 substr $self->{wbuf}, 0, $len, "";
527 410
529 $self->{ww} = AE::io $self->{fh}, 1, $self->{ww_cb} 412 $self->{ww} = AE::io $self->{fh}, 1, $self->{ww_cb}
530 if length $self->{wbuf}; 413 if length $self->{wbuf};
531 } 414 }
532} 415}
533 416
417=item $dbh->attr ($attr_name[, $attr_value], $cb->($dbh, $new_value))
418
419An accessor for the database handle attributes, such as C<AutoCommit>,
420C<RaiseError>, C<PrintError> and so on. If you provide an C<$attr_value>
421(which might be C<undef>), then the given attribute will be set to that
422value.
423
424The callback will be passed the database handle and the attribute's value
425if successful.
426
427If an error occurs and the C<on_error> callback returns, then only C<$dbh>
428will be passed and C<$@> contains the error message.
429
534=item $dbh->exec ("statement", @args, $cb->($dbh, \@rows, $rv)) 430=item $dbh->exec ("statement", @args, $cb->($dbh, \@rows, $rv))
535 431
536Executes the given SQL statement with placeholders replaced by 432Executes the given SQL statement with placeholders replaced by
537C<@args>. The statement will be prepared and cached on the server side, so 433C<@args>. The statement will be prepared and cached on the server side, so
538using placeholders is extremely important. 434using placeholders is extremely important.
545call. 441call.
546 442
547If an error occurs and the C<on_error> callback returns, then only C<$dbh> 443If an error occurs and the C<on_error> callback returns, then only C<$dbh>
548will be passed and C<$@> contains the error message. 444will be passed and C<$@> contains the error message.
549 445
550=item $dbh->attr ($attr_name[, $attr_value], $cb->($dbh, $new_value)) 446=item $dbh->stattr ($attr_name, $cb->($dbh, $value))
551 447
552An accessor for the handle attributes, such as C<AutoCommit>, 448An accessor for the statement attributes of the most recently executed
553C<RaiseError>, C<PrintError> and so on. If you provide an C<$attr_value> 449statement, such as C<NAME> or C<TYPE>.
554(which might be C<undef>), then the given attribute will be set to that
555value.
556 450
557The callback will be passed the database handle and the attribute's value 451The callback will be passed the database handle and the attribute's value
558if successful. 452if successful.
559 453
560If an error occurs and the C<on_error> callback returns, then only C<$dbh> 454If an error occurs and the C<on_error> callback returns, then only C<$dbh>
610 504
611 # otherwise, we can ignore $rc and $errcode for this particular func 505 # otherwise, we can ignore $rc and $errcode for this particular func
612 506
613=cut 507=cut
614 508
615for my $cmd_name (qw(exec attr begin_work commit rollback func)) { 509for my $cmd_name (qw(attr exec stattr begin_work commit rollback func)) {
616 eval 'sub ' . $cmd_name . '{ 510 eval 'sub ' . $cmd_name . '{
617 my $cb = pop; 511 my $cb = pop;
618 splice @_, 1, 0, $cb, (caller)[1,2], "req_' . $cmd_name . '"; 512 splice @_, 1, 0, $cb, (caller)[1,2], "req_' . $cmd_name . '";
619 &_req 513 &_req
620 }'; 514 }';
624 518
625=head1 SEE ALSO 519=head1 SEE ALSO
626 520
627L<AnyEvent>, L<DBI>, L<Coro::Mysql>. 521L<AnyEvent>, L<DBI>, L<Coro::Mysql>.
628 522
629=head1 AUTHOR 523=head1 AUTHOR AND CONTACT
630 524
631 Marc Lehmann <schmorp@schmorp.de> 525 Marc Lehmann <schmorp@schmorp.de> (current maintainer)
632 http://home.schmorp.de/ 526 http://home.schmorp.de/
633 527
634 Adam Rosenstein <adam@redcondor.com> 528 Adam Rosenstein <adam@redcondor.com>
635 http://www.redcondor.com/ 529 http://www.redcondor.com/
636 530
637=cut 531=cut
638 532
6391; 5331
640

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines