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.14 by root, Sat Oct 30 20:23:44 2010 UTC vs.
Revision 1.19 by root, Mon Sep 4 11:46:30 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.1'; 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 or die [$DBI::errstr]]
114}
115
116sub req_commit {
117 [1, $DBH->commit or die [$DBI::errstr]]
118}
119
120sub req_rollback {
121 [1, $DBH->rollback or 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 $self->{last_activity} = AE::now;
310
311 my $len = sysread $client, $rbuf, 65536, length $rbuf;
312 222
313 if ($len > 0) { 223 if ($len > 0) {
314 # we received data, so reset the timer 224 # we received data, so reset the timer
225 $self->{last_activity} = AE::now;
315 226
316 while () { 227 for my $res ($cbor->incr_parse_multiple ($rbuf)) {
317 my $len = unpack "L", $rbuf;
318
319 # full response available?
320 last unless $len && $len + 4 <= length $rbuf;
321
322 my $res = Storable::thaw substr $rbuf, 4;
323 substr $rbuf, 0, $len + 4, ""; # remove length + request
324
325 last unless $self; 228 last unless $self;
229
326 my $req = shift @{ $self->{queue} }; 230 my $req = shift @{ $self->{queue} };
327 231
328 if (defined $res->[0]) { 232 if (defined $res->[0]) {
329 $res->[0] = $self; 233 $res->[0] = $self;
330 $req->[0](@$res); 234 $req->[0](@$res);
… …
335 $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
336 if $self; # cb() could have deleted it 240 if $self; # cb() could have deleted it
337 } 241 }
338 242
339 # no more queued requests, so become idle 243 # no more queued requests, so become idle
244 if ($self && !@{ $self->{queue} }) {
340 undef $self->{last_activity} 245 undef $self->{last_activity};
341 if $self && !@{ $self->{queue} }; 246 $self->{tw_cb}->();
247 }
342 } 248 }
343 249
344 } elsif (defined $len) { 250 } elsif (defined $len) {
345 # todo, caller? 251 # todo, caller?
346 $self->_error ("unexpected eof", @caller, 1); 252 $self->_error ("unexpected eof", @caller, 1);
… …
369 undef $self->{tw}; 275 undef $self->{tw};
370 } 276 }
371 }; 277 };
372 278
373 $self->{ww_cb} = sub { 279 $self->{ww_cb} = sub {
374 return unless $self;
375
376 $self->{last_activity} = AE::now; 280 $self->{last_activity} = AE::now;
377 281
378 my $len = syswrite $client, $self->{wbuf} 282 my $len = syswrite $client, $self->{wbuf}
379 or return delete $self->{ww}; 283 or return delete $self->{ww};
380 284
381 substr $self->{wbuf}, 0, $len, ""; 285 substr $self->{wbuf}, 0, $len, "";
382 }; 286 };
383 } 287 }
384 288
385 my $pid = fork;
386
387 if ($pid) {
388 # parent
389 close $server;
390 } elsif (defined $pid) {
391 # child
392 my $serv_fno = fileno $server;
393
394 if ($self->{exec_server}) {
395 fcntl $server, &Fcntl::F_SETFD, 0; # don't close the server side
396 exec {$^X}
397 "$0 dbi slave",
398 -e => "require shift; AnyEvent::DBI::serve_fd ($serv_fno, $VERSION)",
399 $INC{"AnyEvent/DBI.pm"};
400 POSIX::_exit 124;
401 } else {
402 ($_ != $serv_fno) && POSIX::close $_
403 for $^F+1..$FD_MAX;
404 serve_fh $server, $VERSION;
405
406 # no other way on the broken windows platform, even this leaks
407 # memory and might fail.
408 kill 9, $$
409 if AnyEvent::WIN32;
410
411 # and this kills the parent process on windows
412 POSIX::_exit 0;
413 }
414 } else {
415 croak "fork: $!";
416 }
417
418 $self->{child_pid} = $pid;
419
420 $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;
421 ($self->{on_connect} ? $self->{on_connect} : sub { }), 301 &{ $self->{on_connect} } if $self->{on_connect};
302 },
422 (caller)[1,2], 303 (caller)[1,2],
423 req_open => $dbi, $user, $pass, %dbi_args 304 req_open => $dbi, $user, $pass, %dbi_args
424 ); 305 );
425 306
426 $self 307 $self
… …
432 313
433sub kill_child { 314sub kill_child {
434 my $self = shift; 315 my $self = shift;
435 316
436 if (my $pid = delete $self->{child_pid}) { 317 if (my $pid = delete $self->{child_pid}) {
318 # kill and reap process
319 my $kid_watcher; $kid_watcher = AE::child $pid, sub {
320 undef $kid_watcher;
321 };
437 kill TERM => $pid; 322 kill TERM => $pid;
438 } 323 }
324
325 delete $self->{rw};
326 delete $self->{ww};
327 delete $self->{tw};
439 close delete $self->{fh}; 328 close delete $self->{fh};
440} 329}
441 330
442sub DESTROY { 331sub DESTROY {
443 shift->kill_child; 332 shift->kill_child;
… …
511 if ($self->{timeout} && !$self->{tw}) { 400 if ($self->{timeout} && !$self->{tw}) {
512 $self->{last_activity} = AE::now; 401 $self->{last_activity} = AE::now;
513 $self->{tw_cb}->(); 402 $self->{tw_cb}->();
514 } 403 }
515 404
516 $self->{wbuf} .= pack "L/a*", Storable::freeze \@_; 405 $self->{wbuf} .= CBOR::XS::encode_cbor \@_;
517 406
518 unless ($self->{ww}) { 407 unless ($self->{ww}) {
519 my $len = syswrite $self->{fh}, $self->{wbuf}; 408 my $len = syswrite $self->{fh}, $self->{wbuf};
520 substr $self->{wbuf}, 0, $len, ""; 409 substr $self->{wbuf}, 0, $len, "";
521 410
… …
523 $self->{ww} = AE::io $self->{fh}, 1, $self->{ww_cb} 412 $self->{ww} = AE::io $self->{fh}, 1, $self->{ww_cb}
524 if length $self->{wbuf}; 413 if length $self->{wbuf};
525 } 414 }
526} 415}
527 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
528=item $dbh->exec ("statement", @args, $cb->($dbh, \@rows, $rv)) 430=item $dbh->exec ("statement", @args, $cb->($dbh, \@rows, $rv))
529 431
530Executes the given SQL statement with placeholders replaced by 432Executes the given SQL statement with placeholders replaced by
531C<@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
532using placeholders is extremely important. 434using placeholders is extremely important.
… …
539call. 441call.
540 442
541If 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>
542will be passed and C<$@> contains the error message. 444will be passed and C<$@> contains the error message.
543 445
544=item $dbh->attr ($attr_name[, $attr_value], $cb->($dbh, $new_value)) 446=item $dbh->stattr ($attr_name, $cb->($dbh, $value))
545 447
546An accessor for the handle attributes, such as C<AutoCommit>, 448An accessor for the statement attributes of the most recently executed
547C<RaiseError>, C<PrintError> and so on. If you provide an C<$attr_value> 449statement, such as C<NAME> or C<TYPE>.
548(which might be C<undef>), then the given attribute will be set to that
549value.
550 450
551The 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
552if successful. 452if successful.
553 453
554If 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>
… …
604 504
605 # otherwise, we can ignore $rc and $errcode for this particular func 505 # otherwise, we can ignore $rc and $errcode for this particular func
606 506
607=cut 507=cut
608 508
609for my $cmd_name (qw(exec attr begin_work commit rollback func)) { 509for my $cmd_name (qw(attr exec stattr begin_work commit rollback func)) {
610 eval 'sub ' . $cmd_name . '{ 510 eval 'sub ' . $cmd_name . '{
611 my $cb = pop; 511 my $cb = pop;
612 splice @_, 1, 0, $cb, (caller)[1,2], "req_' . $cmd_name . '"; 512 splice @_, 1, 0, $cb, (caller)[1,2], "req_' . $cmd_name . '";
613 &_req 513 &_req
614 }'; 514 }';
… …
618 518
619=head1 SEE ALSO 519=head1 SEE ALSO
620 520
621L<AnyEvent>, L<DBI>, L<Coro::Mysql>. 521L<AnyEvent>, L<DBI>, L<Coro::Mysql>.
622 522
623=head1 AUTHOR 523=head1 AUTHOR AND CONTACT
624 524
625 Marc Lehmann <schmorp@schmorp.de> 525 Marc Lehmann <schmorp@schmorp.de> (current maintainer)
626 http://home.schmorp.de/ 526 http://home.schmorp.de/
627 527
628 Adam Rosenstein <adam@redcondor.com> 528 Adam Rosenstein <adam@redcondor.com>
629 http://www.redcondor.com/ 529 http://www.redcondor.com/
630 530
631=cut 531=cut
632 532
6331; 5331
634

Diff Legend

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