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

Comparing IO-AIO/AIO.pm (file contents):
Revision 1.55 by root, Sun Oct 22 00:49:29 2006 UTC vs.
Revision 1.63 by root, Mon Oct 23 00:34:36 2006 UTC

15 15
16 aio_read $fh, 30000, 1024, $buffer, 0, sub { 16 aio_read $fh, 30000, 1024, $buffer, 0, sub {
17 $_[0] > 0 or die "read error: $!"; 17 $_[0] > 0 or die "read error: $!";
18 }; 18 };
19 19
20 use IO::AIO 2; # version has aio objects 20 # version 2+ has request and group objects
21 use IO::AIO 2;
21 22
22 my $req = aio_unlink "/tmp/file", sub { }; 23 my $req = aio_unlink "/tmp/file", sub { };
23 $req->cancel; # cancel request if still in queue 24 $req->cancel; # cancel request if still in queue
24 25
25 # AnyEvent 26 my $grp = aio_group sub { print "all stats done\n" };
27 add $grp aio_stat "..." for ...;
28
29 # AnyEvent integration
26 open my $fh, "<&=" . IO::AIO::poll_fileno or die "$!"; 30 open my $fh, "<&=" . IO::AIO::poll_fileno or die "$!";
27 my $w = AnyEvent->io (fh => $fh, poll => 'r', cb => sub { IO::AIO::poll_cb }); 31 my $w = AnyEvent->io (fh => $fh, poll => 'r', cb => sub { IO::AIO::poll_cb });
28 32
29 # Event 33 # Event integration
30 Event->io (fd => IO::AIO::poll_fileno, 34 Event->io (fd => IO::AIO::poll_fileno,
31 poll => 'r', 35 poll => 'r',
32 cb => \&IO::AIO::poll_cb); 36 cb => \&IO::AIO::poll_cb);
33 37
34 # Glib/Gtk2 38 # Glib/Gtk2 integration
35 add_watch Glib::IO IO::AIO::poll_fileno, 39 add_watch Glib::IO IO::AIO::poll_fileno,
36 in => sub { IO::AIO::poll_cb; 1 }; 40 in => sub { IO::AIO::poll_cb; 1 };
37 41
38 # Tk 42 # Tk integration
39 Tk::Event::IO->fileevent (IO::AIO::poll_fileno, "", 43 Tk::Event::IO->fileevent (IO::AIO::poll_fileno, "",
40 readable => \&IO::AIO::poll_cb); 44 readable => \&IO::AIO::poll_cb);
41 45
42 # Danga::Socket 46 # Danga::Socket integration
43 Danga::Socket->AddOtherFds (IO::AIO::poll_fileno => 47 Danga::Socket->AddOtherFds (IO::AIO::poll_fileno =>
44 \&IO::AIO::poll_cb); 48 \&IO::AIO::poll_cb);
45
46 49
47=head1 DESCRIPTION 50=head1 DESCRIPTION
48 51
49This module implements asynchronous I/O using whatever means your 52This module implements asynchronous I/O using whatever means your
50operating system supports. 53operating system supports.
76 our $VERSION = '2.0'; 79 our $VERSION = '2.0';
77 80
78 our @EXPORT = qw(aio_sendfile aio_read aio_write aio_open aio_close aio_stat 81 our @EXPORT = qw(aio_sendfile aio_read aio_write aio_open aio_close aio_stat
79 aio_lstat aio_unlink aio_rmdir aio_readdir aio_scandir aio_symlink 82 aio_lstat aio_unlink aio_rmdir aio_readdir aio_scandir aio_symlink
80 aio_fsync aio_fdatasync aio_readahead aio_rename aio_link aio_move 83 aio_fsync aio_fdatasync aio_readahead aio_rename aio_link aio_move
81 aio_group); 84 aio_group aio_nop);
82 our @EXPORT_OK = qw(poll_fileno poll_cb min_parallel max_parallel max_outstanding nreqs); 85 our @EXPORT_OK = qw(poll_fileno poll_cb min_parallel max_parallel max_outstanding nreqs);
83 86
84 @IO::AIO::GRP::ISA = 'IO::AIO::REQ'; 87 @IO::AIO::GRP::ISA = 'IO::AIO::REQ';
85 88
86 require XSLoader; 89 require XSLoader;
179 print "read $_[0] bytes: <$buffer>\n"; 182 print "read $_[0] bytes: <$buffer>\n";
180 }; 183 };
181 184
182=item aio_move $srcpath, $dstpath, $callback->($status) 185=item aio_move $srcpath, $dstpath, $callback->($status)
183 186
187[EXPERIMENTAL due to internal aio_group use]
188
184Try to move the I<file> (directories not supported as either source or 189Try to move the I<file> (directories not supported as either source or
185destination) from C<$srcpath> to C<$dstpath> and call the callback with 190destination) from C<$srcpath> to C<$dstpath> and call the callback with
186the C<0> (error) or C<-1> ok. 191the C<0> (error) or C<-1> ok.
187 192
188This is a composite request that tries to rename(2) the file first. If 193This is a composite request that tries to rename(2) the file first. If
198=cut 203=cut
199 204
200sub aio_move($$$) { 205sub aio_move($$$) {
201 my ($src, $dst, $cb) = @_; 206 my ($src, $dst, $cb) = @_;
202 207
203 my $grp = aio_group; 208 my $grp = aio_group $cb;
204 209
205 add $grp aio_rename $src, $dst, sub { 210 add $grp aio_rename $src, $dst, sub {
206 if ($_[0] && $! == EXDEV) { 211 if ($_[0] && $! == EXDEV) {
207 add $grp aio_open $src, O_RDONLY, 0, sub { 212 add $grp aio_open $src, O_RDONLY, 0, sub {
208 if (my $src_fh = $_[0]) { 213 if (my $src_fh = $_[0]) {
218 chmod $stat[2] & 07777, $dst_fh; 223 chmod $stat[2] & 07777, $dst_fh;
219 chown $stat[4], $stat[5], $dst_fh; 224 chown $stat[4], $stat[5], $dst_fh;
220 close $dst_fh; 225 close $dst_fh;
221 226
222 add $grp aio_unlink $src, sub { 227 add $grp aio_unlink $src, sub {
223 $cb->($_[0]); 228 $grp->result ($_[0]);
224 }; 229 };
225 } else { 230 } else {
226 my $errno = $!; 231 my $errno = $!;
227 add $grp aio_unlink $dst, sub { 232 add $grp aio_unlink $dst, sub {
228 $! = $errno; 233 $! = $errno;
229 $cb->(-1); 234 $grp->result (-1);
230 }; 235 };
231 } 236 }
232 }; 237 };
233 } else { 238 } else {
234 $cb->(-1); 239 $grp->result (-1);
235 } 240 }
236 }, 241 },
237 242
238 } else { 243 } else {
239 $cb->(-1); 244 $grp->result (-1);
240 } 245 }
241 }; 246 };
242 } else { 247 } else {
243 $cb->($_[0]); 248 $grp->result ($_[0]);
244 } 249 }
245 }; 250 };
246 251
247 $grp 252 $grp
248} 253}
340The callback a single argument which is either C<undef> or an array-ref 345The callback a single argument which is either C<undef> or an array-ref
341with the filenames. 346with the filenames.
342 347
343=item aio_scandir $path, $maxreq, $callback->($dirs, $nondirs) 348=item aio_scandir $path, $maxreq, $callback->($dirs, $nondirs)
344 349
350[EXPERIMENTAL due to internal aio_group use]
351
345Scans a directory (similar to C<aio_readdir>) but additionally tries to 352Scans a directory (similar to C<aio_readdir>) but additionally tries to
346separate the entries of directory C<$path> into two sets of names, ones 353separate the entries of directory C<$path> into two sets of names, ones
347you can recurse into (directories or links to them), and ones you cannot 354you can recurse into (directories or links to them), and ones you cannot
348recurse into (everything else). 355recurse into (everything else).
349 356
350C<aio_scandir> is a composite request that consists of many sub 357C<aio_scandir> is a composite request that creates of many sub requests_
351requests. C<$maxreq> specifies the maximum number of outstanding aio 358C<$maxreq> specifies the maximum number of outstanding aio requests that
352requests that this function generates. If it is C<< <= 0 >>, then a 359this function generates. If it is C<< <= 0 >>, then a suitable default
353suitable default will be chosen (currently 8). 360will be chosen (currently 6).
354 361
355On error, the callback is called without arguments, otherwise it receives 362On error, the callback is called without arguments, otherwise it receives
356two array-refs with path-relative entry names. 363two array-refs with path-relative entry names.
357 364
358Example: 365Example:
395=cut 402=cut
396 403
397sub aio_scandir($$$) { 404sub aio_scandir($$$) {
398 my ($path, $maxreq, $cb) = @_; 405 my ($path, $maxreq, $cb) = @_;
399 406
400 my $grp = aio_group; 407 my $grp = aio_group $cb;
401 408
402 $maxreq = 8 if $maxreq <= 0; 409 $maxreq = 6 if $maxreq <= 0;
403 410
404 # stat once 411 # stat once
405 add $grp aio_stat $path, sub { 412 add $grp aio_stat $path, sub {
406 return $cb->() if $_[0]; 413 return $grp->result () if $_[0];
407 my $now = time; 414 my $now = time;
408 my $hash1 = join ":", (stat _)[0,1,3,7,9]; 415 my $hash1 = join ":", (stat _)[0,1,3,7,9];
409 416
410 # read the directory entries 417 # read the directory entries
411 add $grp aio_readdir $path, sub { 418 add $grp aio_readdir $path, sub {
412 my $entries = shift 419 my $entries = shift
413 or return $cb->(); 420 or return $grp->result ();
414 421
415 # stat the dir another time 422 # stat the dir another time
416 add $grp aio_stat $path, sub { 423 add $grp aio_stat $path, sub {
417 my $hash2 = join ":", (stat _)[0,1,3,7,9]; 424 my $hash2 = join ":", (stat _)[0,1,3,7,9];
418 425
423 $ndirs = -1; 430 $ndirs = -1;
424 } else { 431 } else {
425 # if nlink == 2, we are finished 432 # if nlink == 2, we are finished
426 # on non-posix-fs's, we rely on nlink < 2 433 # on non-posix-fs's, we rely on nlink < 2
427 $ndirs = (stat _)[3] - 2 434 $ndirs = (stat _)[3] - 2
428 or return $cb->([], $entries); 435 or return $grp->result ([], $entries);
429 } 436 }
430 437
431 # sort into likely dirs and likely nondirs 438 # sort into likely dirs and likely nondirs
432 # dirs == files without ".", short entries first 439 # dirs == files without ".", short entries first
433 $entries = [map $_->[0], 440 $entries = [map $_->[0],
438 my (@dirs, @nondirs); 445 my (@dirs, @nondirs);
439 446
440 my ($statcb, $schedcb); 447 my ($statcb, $schedcb);
441 my $nreq = 0; 448 my $nreq = 0;
442 449
450 my $statgrp = add $grp aio_group;
451
443 $schedcb = sub { 452 $schedcb = sub {
444 if (@$entries) { 453 if (@$entries) {
445 if ($nreq < $maxreq) { 454 if ($nreq < $maxreq) {
446 my $ent = pop @$entries; 455 my $ent = pop @$entries;
447 $nreq++; 456 $nreq++;
448 add $grp aio_stat "$path/$ent/.", sub { $statcb->($_[0], $ent) }; 457 add $statgrp aio_stat "$path/$ent/.", sub { $statcb->($_[0], $ent) };
449 } 458 }
450 } elsif (!$nreq) { 459 } elsif (!$nreq) {
451 # finished 460 # finished
461 $statgrp->cancel;
452 undef $statcb; 462 undef $statcb;
453 undef $schedcb; 463 undef $schedcb;
454 $cb->(\@dirs, \@nondirs) if $cb; 464 $grp->result (\@dirs, \@nondirs);
455 undef $cb;
456 } 465 }
457 }; 466 };
458 $statcb = sub { 467 $statcb = sub {
459 my ($status, $entry) = @_; 468 my ($status, $entry) = @_;
460 469
502callback with the fdatasync result code. 511callback with the fdatasync result code.
503 512
504If this call isn't available because your OS lacks it or it couldn't be 513If this call isn't available because your OS lacks it or it couldn't be
505detected, it will be emulated by calling C<fsync> instead. 514detected, it will be emulated by calling C<fsync> instead.
506 515
507=item aio_group $callback->() 516=item aio_group $callback->(...)
508 517
509[EXPERIMENTAL] 518[EXPERIMENTAL]
510 519
511This is a very special aio request: Instead of doing something, it is a 520This is a very special aio request: Instead of doing something, it is a
512container for other aio requests, which is useful if you want to bundle 521container for other aio requests, which is useful if you want to bundle
524 add $grp 533 add $grp
525 (aio_stat ...), 534 (aio_stat ...),
526 (aio_stat ...), 535 (aio_stat ...),
527 ...; 536 ...;
528 537
538=item aio_nop $callback->()
539
540This is a special request - it does nothing in itself and is only used for
541side effects, such as when you want to add a dummy request to a group so
542that finishing the requests in the group depends on executing the given
543code.
544
529=item aio_sleep $fractional_seconds, $callback->() *NOT EXPORTED* 545=item IO::AIO::aio_sleep $fractional_seconds, $callback->() *NOT EXPORTED*
530 546
531Mainly used for debugging and benchmarking, this aio request puts one of 547Mainly used for debugging and benchmarking, this aio request puts one of
532the request workers to sleep for the given time. 548the request workers to sleep for the given time.
549
550While it is theoretically handy to have simple I/O scheduling requests
551like sleep and file handle readable/writable, the overhead this creates
552is immense, so do not use this function except to put your application
553under artificial I/O pressure.
533 554
534=back 555=back
535 556
536=head2 IO::AIO::REQ CLASS 557=head2 IO::AIO::REQ CLASS
537 558
577You add requests by calling the C<add> method with one or more 598You add requests by calling the C<add> method with one or more
578C<IO::AIO::REQ> objects: 599C<IO::AIO::REQ> objects:
579 600
580 $grp->add (aio_unlink "..."); 601 $grp->add (aio_unlink "...");
581 602
582 add $grp aio_stat "...", sub { ... }; 603 add $grp aio_stat "...", sub {
604 $_[0] or return $grp->result ("error");
605
606 # add another request dynamically, if first succeeded
607 add $grp aio_open "...", sub {
608 $grp->result ("ok");
609 };
610 };
583 611
584This makes it very easy to create composite requests (see the source of 612This makes it very easy to create composite requests (see the source of
585C<aio_move> for an application) that work and feel like simple requests. 613C<aio_move> for an application) that work and feel like simple requests.
586 614
615=over 4
616
587The IO::AIO::GRP objects will be cleaned up during calls to 617=item * The IO::AIO::GRP objects will be cleaned up during calls to
588C<IO::AIO::poll_cb>, just like any other request. 618C<IO::AIO::poll_cb>, just like any other request.
589 619
590They can be canceled like any other request. Canceling will cancel not 620=item * They can be canceled like any other request. Canceling will cancel not
591just the request itself, but also all requests it contains. 621only the request itself, but also all requests it contains.
592 622
593They can also can also be added to other IO::AIO::GRP objects. 623=item * They can also can also be added to other IO::AIO::GRP objects.
624
625=item * You must not add requests to a group from within the group callback (or
626any later time).
627
628=item * This does not harmonise well with C<max_outstanding>, so best do
629not combine C<aio_group> with it. Groups and feeders are recommended for
630this kind of concurrency-limiting.
631
632=back
594 633
595Their lifetime, simplified, looks like this: when they are empty, they 634Their lifetime, simplified, looks like this: when they are empty, they
596will finish very quickly. If they contain only requests that are in the 635will finish very quickly. If they contain only requests that are in the
597C<done> state, they will also finish. Otherwise they will continue to 636C<done> state, they will also finish. Otherwise they will continue to
598exist. 637exist.
599 638
639That means after creating a group you have some time to add requests. And
640in the callbacks of those requests, you can add further requests to the
641group. And only when all those requests have finished will the the group
642itself finish.
643
600=over 4 644=over 4
601 645
602=item $grp->add (...) 646=item $grp->add (...)
603 647
604=item add $grp ... 648=item add $grp ...
605 649
606Add one or more 650Add one or more requests to the group. Any type of L<IO::AIO::REQ> can
607Cancels the request, if possible. Has the effect of skipping execution 651be added, including other groups, as long as you do not create circular
608when entering the B<execute> state and skipping calling the callback when 652dependencies.
609entering the the B<result> state, but will leave the request otherwise 653
610untouched. That means that requests that currently execute will not be 654Returns all its arguments.
611stopped and resources held by the request will not be freed prematurely. 655
656=item $grp->result (...)
657
658Set the result value(s) that will be passed to the group callback when all
659subrequests have finished. By default, no argument will be passed.
660
661=item $grp->set_feeder ($callback->($grp))
662
663[VERY EXPERIMENTAL]
664
665Sets a feeder/generator on this group: every group can have an attached
666generator that generates requests if idle. The idea behind this is that,
667although you could just queue as many requests as you want in a group,
668this might starve other requests for a potentially long time. For
669example, C<aio_scandir> might generate hundreds of thousands C<aio_stat>
670requests, delaying any later requests for a long time.
671
672To avoid this, and allow incremental generation of requests, you can
673instead a group and set a feeder on it that generates those requests. The
674feeder will be called whenever there are few enough (see C<feeder_limit>,
675below) requests active in the group itself and is expected to queue more
676requests.
677
678The feeder can queue as many requests as it likes (i.e. C<add> does not
679impose any limits).
680
681If the feeder does not queue more requests when called, it will be
682automatically removed from the group.
683
684If the feeder limit is C<0>, it will be set to C<2> automatically.
685
686Example:
687
688 # stat all files in @files, but only ever use four aio requests concurrently:
689
690 my $grp = aio_group sub { print "finished\n" };
691 $grp->feeder_limit (4);
692 $grp->set_feeder (sub {
693 my $file = pop @files
694 or return;
695
696 add $grp aio_stat $file, sub { ... };
697 });
698
699=item $grp->feeder_limit ($num)
700
701Sets the feeder limit for the group: The feeder will be called whenever
702the group contains less than this many requests.
703
704Setting the limit to C<0> will pause the feeding process.
612 705
613=back 706=back
614 707
615=head2 SUPPORT FUNCTIONS 708=head2 SUPPORT FUNCTIONS
616 709
674 IO::AIO::poll_wait, IO::AIO::poll_cb 767 IO::AIO::poll_wait, IO::AIO::poll_cb
675 if IO::AIO::nreqs; 768 if IO::AIO::nreqs;
676 769
677=item IO::AIO::min_parallel $nthreads 770=item IO::AIO::min_parallel $nthreads
678 771
679Set the minimum number of AIO threads to C<$nthreads>. The current default 772Set the minimum number of AIO threads to C<$nthreads>. The current
680is C<4>, which means four asynchronous operations can be done at one time 773default is C<8>, which means eight asynchronous operations can execute
681(the number of outstanding operations, however, is unlimited). 774concurrently at any one time (the number of outstanding requests,
775however, is unlimited).
682 776
683IO::AIO starts threads only on demand, when an AIO request is queued and 777IO::AIO starts threads only on demand, when an AIO request is queued and
684no free thread exists. 778no free thread exists.
685 779
686It is recommended to keep the number of threads low, as some Linux 780It is recommended to keep the number of threads relatively low, as some
687kernel versions will scale negatively with the number of threads (higher 781Linux kernel versions will scale negatively with the number of threads
688parallelity => MUCH higher latency). With current Linux 2.6 versions, 4-32 782(higher parallelity => MUCH higher latency). With current Linux 2.6
689threads should be fine. 783versions, 4-32 threads should be fine.
690 784
691Under most circumstances you don't need to call this function, as the 785Under most circumstances you don't need to call this function, as the
692module selects a default that is suitable for low to moderate load. 786module selects a default that is suitable for low to moderate load.
693 787
694=item IO::AIO::max_parallel $nthreads 788=item IO::AIO::max_parallel $nthreads
705 799
706Under normal circumstances you don't need to call this function. 800Under normal circumstances you don't need to call this function.
707 801
708=item $oldnreqs = IO::AIO::max_outstanding $nreqs 802=item $oldnreqs = IO::AIO::max_outstanding $nreqs
709 803
804[DEPRECATED]
805
710Sets the maximum number of outstanding requests to C<$nreqs>. If you 806Sets the maximum number of outstanding requests to C<$nreqs>. If you
711try to queue up more than this number of requests, the caller will block until 807try to queue up more than this number of requests, the caller will block until
712some requests have been handled. 808some requests have been handled.
713 809
714The default is very large, so normally there is no practical limit. If you 810The default is very large, so normally there is no practical limit. If you
715queue up many requests in a loop it often improves speed if you set 811queue up many requests in a loop it often improves speed if you set
716this to a relatively low number, such as C<100>. 812this to a relatively low number, such as C<100>.
813
814This function does not work well together with C<aio_group>'s, and their
815feeder interface is better suited to limiting concurrency, so do not use
816this function.
717 817
718Under normal circumstances you don't need to call this function. 818Under normal circumstances you don't need to call this function.
719 819
720=back 820=back
721 821
735 or return undef; 835 or return undef;
736 836
737 *$sym 837 *$sym
738} 838}
739 839
740min_parallel 4; 840min_parallel 8;
741 841
742END { 842END {
743 max_parallel 0; 843 max_parallel 0;
744} 844}
745 845
759 859
760In short: the parent will, after a short pause, continue as if fork had 860In short: the parent will, after a short pause, continue as if fork had
761not been called, while the child will act as if IO::AIO has not been used 861not been called, while the child will act as if IO::AIO has not been used
762yet. 862yet.
763 863
864=head2 MEMORY USAGE
865
866Each aio request uses - depending on your architecture - around 128 bytes
867of memory. In addition, stat requests need a stat buffer (possibly a few
868hundred bytes). Perl scalars and other data passed into aio requests will
869also be locked.
870
871This is now awfully much, so queuing lots of requests is not usually a
872problem.
873
874Each thread needs a stack area which is usually around 16k, sometimes much
875larger, depending on the OS.
876
764=head1 SEE ALSO 877=head1 SEE ALSO
765 878
766L<Coro>, L<Linux::AIO> (obsolete). 879L<Coro>, L<Linux::AIO> (obsolete).
767 880
768=head1 AUTHOR 881=head1 AUTHOR

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines