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.13 by root, Sat Oct 23 21:47:13 2010 UTC vs.
Revision 1.18 by root, Sun Aug 27 09:54:25 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 120% to 200% (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 strict qw(vars subs); 61use common::sense;
61no warnings;
62 62
63use Carp; 63use Carp;
64use Socket (); 64use Convert::Scalar ();
65use Scalar::Util (); 65use AnyEvent::Fork ();
66use Storable (); 66use CBOR::XS ();
67
68use DBI ();
69 67
70use AnyEvent (); 68use AnyEvent ();
71use AnyEvent::Util (); 69use AnyEvent::Util ();
72 70
73use Errno (); 71use Errno ();
74use Fcntl ();
75use POSIX ();
76 72
77our $VERSION = '2.0'; 73our $VERSION = '3.01';
78
79our $FD_MAX = eval { POSIX::sysconf (&POSIX::_SC_OPEN_MAX) - 1 } || 1023;
80
81# this is the forked server code, could/should be bundled as it's own file
82
83our $DBH;
84
85sub req_open {
86 my (undef, $dbi, $user, $pass, %attr) = @{+shift};
87
88 $DBH = DBI->connect ($dbi, $user, $pass, \%attr) or die $DBI::errstr;
89
90 [1, 1]
91}
92
93sub req_exec {
94 my (undef, $st, @args) = @{+shift};
95 my $sth = $DBH->prepare_cached ($st, undef, 1)
96 or die [$DBI::errstr];
97
98 my $rv = $sth->execute (@args)
99 or die [$sth->errstr];
100
101 [1, $sth->{NUM_OF_FIELDS} ? $sth->fetchall_arrayref : undef, $rv]
102}
103
104sub req_attr {
105 my (undef, $attr_name, @attr_val) = @{+shift};
106
107 $DBH->{$attr_name} = $attr_val[0]
108 if @attr_val;
109
110 [1, $DBH->{$attr_name}]
111}
112
113sub req_begin_work {
114 [1, $DBH->begin_work or die [$DBI::errstr]]
115}
116
117sub req_commit {
118 [1, $DBH->commit or die [$DBI::errstr]]
119}
120
121sub req_rollback {
122 [1, $DBH->rollback or die [$DBI::errstr]]
123}
124
125sub req_func {
126 my (undef, $arg_string, $function) = @{+shift};
127 my @args = eval $arg_string;
128
129 die "error evaling \$dbh->func() arg_string: $@"
130 if $@;
131
132 my $rc = $DBH->func (@args, $function);
133 return [1, $rc, $DBI::err, $DBI::errstr];
134}
135
136sub serve_fh($$) {
137 my ($fh, $version) = @_;
138
139 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
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 $wbuf = pack "L/a*", Storable::freeze [undef, ref $@ ? ("$@->[0]", $@->[1]) : ("$@", 1)]
165 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}
175
176sub serve_fd($$) {
177 open my $fh, ">>&=$_[0]"
178 or die "Couldn't open server file descriptor: $!";
179
180 serve_fh $fh, $_[1];
181}
182 74
183=head2 METHODS 75=head2 METHODS
184 76
185=over 4 77=over 4
186 78
… …
230 122
231When 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
232will 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
233received from C<new> to make requests. 125received from C<new> to make requests.
234 126
235=item exec_server => 1 127=item fork_template => $AnyEvent::Fork-object
236 128
237If 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
238fork 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
239AnyEvent::DBI proxy running. This will provide the cleanest possible proxy 131C<exec_server> constructor argument) or uses a process forked early (see
240for your database server. 132L<AnyEvent::Fork::Early>).
241 133
242If 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
243false 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
244the 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
245itself up by calling POSIX::close on all file descriptors except STDIN, 137and again into your child processes:
246STDOUT, and STDERR (and the socket it uses to communicate with the cilent, 138
247of 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;
248 146
249=item timeout => seconds 147=item timeout => seconds
250 148
251If you supply a timeout parameter (fractional values are supported), then 149If you supply a timeout parameter (fractional values are supported), then
252a 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
… …
267 165
268=back 166=back
269 167
270Any additional key-value pairs will be rolled into a hash reference 168Any additional key-value pairs will be rolled into a hash reference
271and passed as the final argument to the C<< DBI->connect (...) >> 169and passed as the final argument to the C<< DBI->connect (...) >>
272call. 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
273AnyEvent::Handle you could do: 171AnyEvent::Handle you could do:
274 172
275 $dbh = new AnyEvent::DBI 173 $dbh = new AnyEvent::DBI
276 "DBI:mysql:test;mysql_read_default_file=/root/.my.cnf", "", "", 174 "DBI:mysql:test;mysql_read_default_file=/root/.my.cnf", "", "",
277 PrintError => 0, 175 PrintError => 0,
… …
279 $log_handle->push_write ("DBI Error: $@ at $_[1]:$_[2]\n"); 177 $log_handle->push_write ("DBI Error: $@ at $_[1]:$_[2]\n");
280 }; 178 };
281 179
282=cut 180=cut
283 181
284# stupid Storable autoloading, total loss-loss situation
285Storable::thaw Storable::freeze [];
286
287sub new { 182sub new {
288 my ($class, $dbi, $user, $pass, %arg) = @_; 183 my ($class, $dbi, $user, $pass, %arg) = @_;
289 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.
290 my ($client, $server) = AnyEvent::Util::portable_socketpair 189 my ($client, $server) = AnyEvent::Util::portable_socketpair
291 or croak "unable to create Anyevent::DBI communications pipe: $!"; 190 or croak "unable to create AnyEvent::DBI communications pipe: $!";
191
192 my $fork = delete $arg{fork_template};
292 193
293 my %dbi_args = %arg; 194 my %dbi_args = %arg;
294 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)};
295 196
296 my $self = bless \%arg, $class; 197 my $self = bless \%arg, $class;
198
297 $self->{fh} = $client; 199 $self->{fh} = $client;
298
299 AnyEvent::Util::fh_nonblocking $client, 1;
300 200
301 my $rbuf; 201 my $rbuf;
302 my @caller = (caller)[1,2]; # the "default" caller 202 my @caller = (caller)[1,2]; # the "default" caller
303 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
304 { 215 {
305 Scalar::Util::weaken (my $self = $self); 216 Convert::Scalar::weaken (my $self = $self);
306 217
307 $self->{rw} = AnyEvent->io (fh => $client, poll => "r", cb => sub { 218 my $cbor = new CBOR::XS;
308 return unless $self;
309 219
310 $self->{last_activity} = AnyEvent->now; 220 $self->{rw} = AE::io $client, 0, sub {
311 221 my $len = Convert::Scalar::extend_read $client, $rbuf, 65536;
312 my $len = sysread $client, $rbuf, 65536, length $rbuf;
313 222
314 if ($len > 0) { 223 if ($len > 0) {
315 # we received data, so reset the timer 224 # we received data, so reset the timer
225 $self->{last_activity} = AE::now;
316 226
317 while () { 227 for my $res ($cbor->incr_parse_multiple ($rbuf)) {
318 my $len = unpack "L", $rbuf;
319
320 # full response available?
321 last unless $len && $len + 4 <= length $rbuf;
322
323 my $res = Storable::thaw substr $rbuf, 4;
324 substr $rbuf, 0, $len + 4, ""; # remove length + request
325
326 last unless $self; 228 last unless $self;
229
327 my $req = shift @{ $self->{queue} }; 230 my $req = shift @{ $self->{queue} };
328 231
329 if (defined $res->[0]) { 232 if (defined $res->[0]) {
330 $res->[0] = $self; 233 $res->[0] = $self;
331 $req->[0](@$res); 234 $req->[0](@$res);
… …
336 $self->_error ($res->[1], @$req, $res->[2]) # error, request record, is_fatal 239 $self->_error ($res->[1], @$req, $res->[2]) # error, request record, is_fatal
337 if $self; # cb() could have deleted it 240 if $self; # cb() could have deleted it
338 } 241 }
339 242
340 # no more queued requests, so become idle 243 # no more queued requests, so become idle
244 if ($self && !@{ $self->{queue} }) {
341 undef $self->{last_activity} 245 undef $self->{last_activity};
342 if $self && !@{ $self->{queue} }; 246 $self->{tw_cb}->();
247 }
343 } 248 }
344 249
345 } elsif (defined $len) { 250 } elsif (defined $len) {
346 # todo, caller? 251 # todo, caller?
347 $self->_error ("unexpected eof", @caller, 1); 252 $self->_error ("unexpected eof", @caller, 1);
348 } elsif ($! != Errno::EAGAIN) { 253 } elsif ($! != Errno::EAGAIN) {
349 # todo, caller? 254 # todo, caller?
350 $self->_error ("read error: $!", @caller, 1); 255 $self->_error ("read error: $!", @caller, 1);
351 } 256 }
352 }); 257 };
353 258
354 $self->{tw_cb} = sub { 259 $self->{tw_cb} = sub {
355 if ($self->{timeout} && $self->{last_activity}) { 260 if ($self->{timeout} && $self->{last_activity}) {
356 if (AnyEvent->now > $self->{last_activity} + $self->{timeout}) { 261 if (AE::now > $self->{last_activity} + $self->{timeout}) {
357 # we did time out 262 # we did time out
358 my $req = $self->{queue}[0]; 263 my $req = $self->{queue}[0];
359 $self->_error (timeout => $req->[1], $req->[2], 1); # timeouts are always fatal 264 $self->_error (timeout => $req->[1], $req->[2], 1); # timeouts are always fatal
360 } else { 265 } else {
361 # we need to re-set the timeout watcher 266 # we need to re-set the timeout watcher
362 $self->{tw} = AnyEvent->timer ( 267 $self->{tw} = AE::timer
363 after => $self->{last_activity} + $self->{timeout} - AnyEvent->now, 268 $self->{last_activity} + $self->{timeout} - AE::now,
269 0,
364 cb => $self->{tw_cb}, 270 $self->{tw_cb},
365 ); 271 ;
366 Scalar::Util::weaken $self;
367 } 272 }
368 } else { 273 } else {
369 # no timeout check wanted, or idle 274 # no timeout check wanted, or idle
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} = AnyEvent->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
… …
430sub _server_pid { 310sub _server_pid {
431 shift->{child_pid} 311 shift->{child_pid}
432} 312}
433 313
434sub kill_child { 314sub kill_child {
435 my $self = shift; 315 my $self = shift;
316
436 my $child_pid = delete $self->{child_pid}; 317 if (my $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 318 # kill and reap process
447 my $kid_watcher; $kid_watcher = AnyEvent->child ( 319 my $kid_watcher; $kid_watcher = AE::child $pid, sub {
448 pid => $child_pid,
449 cb => sub {
450 # just hold on to this so it won't go away
451 undef $kid_watcher; 320 undef $kid_watcher;
452 # cancel SIGKILL
453 undef $murder_timer;
454 },
455 ); 321 };
322 kill TERM => $pid;
323 }
456 324
325 delete $self->{rw};
326 delete $self->{ww};
327 delete $self->{tw};
457 close $self->{fh}; 328 close delete $self->{fh};
458 }
459} 329}
460 330
461sub DESTROY { 331sub DESTROY {
462 shift->kill_child; 332 shift->kill_child;
463} 333}
… …
526 396
527 push @{ $self->{queue} }, [$cb, $filename, $line]; 397 push @{ $self->{queue} }, [$cb, $filename, $line];
528 398
529 # re-start timeout if necessary 399 # re-start timeout if necessary
530 if ($self->{timeout} && !$self->{tw}) { 400 if ($self->{timeout} && !$self->{tw}) {
531 $self->{last_activity} = AnyEvent->now; 401 $self->{last_activity} = AE::now;
532 $self->{tw_cb}->(); 402 $self->{tw_cb}->();
533 } 403 }
534 404
535 $self->{wbuf} .= pack "L/a*", Storable::freeze \@_; 405 $self->{wbuf} .= CBOR::XS::encode_cbor \@_;
536 406
537 unless ($self->{ww}) { 407 unless ($self->{ww}) {
538 my $len = syswrite $self->{fh}, $self->{wbuf}; 408 my $len = syswrite $self->{fh}, $self->{wbuf};
539 substr $self->{wbuf}, 0, $len, ""; 409 substr $self->{wbuf}, 0, $len, "";
540 410
541 # still any left? then install a write watcher 411 # still any left? then install a write watcher
542 $self->{ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $self->{ww_cb}) 412 $self->{ww} = AE::io $self->{fh}, 1, $self->{ww_cb}
543 if length $self->{wbuf}; 413 if length $self->{wbuf};
544 } 414 }
545} 415}
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.
546 429
547=item $dbh->exec ("statement", @args, $cb->($dbh, \@rows, $rv)) 430=item $dbh->exec ("statement", @args, $cb->($dbh, \@rows, $rv))
548 431
549Executes the given SQL statement with placeholders replaced by 432Executes the given SQL statement with placeholders replaced by
550C<@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
… …
558call. 441call.
559 442
560If 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>
561will be passed and C<$@> contains the error message. 444will be passed and C<$@> contains the error message.
562 445
563=item $dbh->attr ($attr_name[, $attr_value], $cb->($dbh, $new_value)) 446=item $dbh->stattr ($attr_name, $cb->($dbh, $value))
564 447
565An accessor for the handle attributes, such as C<AutoCommit>, 448An accessor for the statement attributes of the most recently executed
566C<RaiseError>, C<PrintError> and so on. If you provide an C<$attr_value> 449statement, such as C<NAME> or C<TYPE>.
567(which might be C<undef>), then the given attribute will be set to that
568value.
569 450
570The 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
571if successful. 452if successful.
572 453
573If 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>
… …
623 504
624 # otherwise, we can ignore $rc and $errcode for this particular func 505 # otherwise, we can ignore $rc and $errcode for this particular func
625 506
626=cut 507=cut
627 508
628for my $cmd_name (qw(exec attr begin_work commit rollback func)) { 509for my $cmd_name (qw(attr exec stattr begin_work commit rollback func)) {
629 eval 'sub ' . $cmd_name . '{ 510 eval 'sub ' . $cmd_name . '{
630 my $cb = pop; 511 my $cb = pop;
631 splice @_, 1, 0, $cb, (caller)[1,2], "req_' . $cmd_name . '"; 512 splice @_, 1, 0, $cb, (caller)[1,2], "req_' . $cmd_name . '";
632 &_req 513 &_req
633 }'; 514 }';
… …
637 518
638=head1 SEE ALSO 519=head1 SEE ALSO
639 520
640L<AnyEvent>, L<DBI>, L<Coro::Mysql>. 521L<AnyEvent>, L<DBI>, L<Coro::Mysql>.
641 522
642=head1 AUTHOR 523=head1 AUTHOR AND CONTACT
643 524
644 Marc Lehmann <schmorp@schmorp.de> 525 Marc Lehmann <schmorp@schmorp.de> (current maintainer)
645 http://home.schmorp.de/ 526 http://home.schmorp.de/
646 527
647 Adam Rosenstein <adam@redcondor.com> 528 Adam Rosenstein <adam@redcondor.com>
648 http://www.redcondor.com/ 529 http://www.redcondor.com/
649 530
650=cut 531=cut
651 532
6521; 5331
653

Diff Legend

– Removed lines
+ Added lines
< Changed lines
> Changed lines