Coro/Coro/AIO.pm

=head1 NAME

Coro::AIO - truly asynchronous file and directrory I/O

=head1 SYNOPSIS

   use Coro::AIO;

   # can now use any of the aio requests your IO::AIO module supports.

   # read 1MB of /etc/passwd, without blocking other coroutines
   my $fh = aio_open "/etc/passwd", O_RDONLY, 0
      or die "/etc/passwd: $!";
   aio_read $fh, 0, 1_000_000, my $buf, 0
      or die "aio_read: $!";
   aio_close $fh;

=head1 DESCRIPTION

This module implements a thin wrapper around L<IO::AIO|IO::AIO>. All of
the functions that expect a callback are being wrapped by this module.

The API is exactly the same as that of the corresponding IO::AIO routines,
except that you have to specify I<all> arguments I<except> the callback
argument. Instead the routines return the values normally passed to the
callback. Everything else, including C<$!> and perls stat cache, are set
as expected after these functions return.

You can mix calls to C<IO::AIO> functions with calls to this module. You
I<must not>, however, call these routines from within IO::AIO callbacks,
as this causes a deadlock. Start a coro inside the callback instead.

You also can, but do not need to, call C<IO::AIO::poll_cb>, as this
module automatically installs an event watcher for the C<IO::AIO> file
descriptor. It uses the L<AnyEvent|AnyEvent> module for this, so please
refer to its documentation on how it selects an appropriate Event module.

All other functions exported by default by IO::AIO (e.g. C<aioreq_pri>)
will be exported by default by Coro::AIO, too.

Functions that can be optionally imported from IO::AIO can be imported
from Coro::AIO or can be called directly, e.g. C<Coro::AIO::nreqs>.

You cannot specify priorities with C<aioreq_pri>, as this module
overwrites the request priority with the current coroutine priority at all
times.

For your convienience, here are the changed function signatures for most
of the requests, for documentation of these functions please have a look
at L<IO::AIO|the IO::AIO manual>.

The AnyEvent watcher can be disabled by executing C<undef
$Coro::AIO::WATCHER>. Please notify the author of when and why you think
this was necessary.

=over 4

=cut

package Coro::AIO;

use strict qw(subs vars);

use Coro ();
use AnyEvent;
use IO::AIO ();

use base Exporter::;

if (AnyEvent::detect =~ /^AnyEvent::Impl::(Coro)?EV$/) {
   $WATCHER = EV::io IO::AIO::poll_fileno, EV::READ, \&IO::AIO::poll_cb;
} else {
   our $FH; open $FH, "<&=" . IO::AIO::poll_fileno;
   $WATCHER = AnyEvent->io (fh => $FH, poll => 'r', cb => \&IO::AIO::poll_cb);
}


our @EXPORT    = @IO::AIO::EXPORT;
our @EXPORT_OK = @IO::AIO::EXPORT_OK;
our $AUTOLOAD;

{
   my @reqs = @IO::AIO::AIO_REQ ? @IO::AIO::AIO_REQ : @EXPORT;
   my %reqs = map +($_ => 1), @reqs;

   eval
      join "",
         map "sub $_(" . (prototype "IO::AIO::$_") . ");",
            grep !$reqs{$_},
                @EXPORT, @EXPORT_OK;

   for my $sub (@reqs) {
      push @EXPORT, $sub;
      
      my $iosub = "IO::AIO::$sub";
      my $proto = prototype $iosub;

      $proto =~ s/;?\$$// or die "$iosub: unable to remove callback slot from prototype";

      eval qq{
#line 1 "Coro::AIO::$sub($proto)"
         sub $sub($proto) {
            my \$current = \$Coro::current;
            my \$state;
            my \@res;

            push \@_, sub {
               \$state = _get_state;
               \@res = \@_;
               \$current->ready;
            };

            aioreq_pri \$Coro::current->prio;
            &$iosub;

            &Coro::schedule;
            &Coro::schedule while !\$state;

            _set_state \$state;
            wantarray ? \@res : \$res[0]
         }
      };
      die if $@;
   }
}

sub AUTOLOAD {
   (my $func = $AUTOLOAD) =~ s/^.*:://;
   *$AUTOLOAD = \&{"IO::AIO::$func"};
   goto &$AUTOLOAD;
}

=item $fh = aio_open $pathname, $flags, $mode

=item $status = aio_close $fh

=item $retval = aio_read  $fh,$offset,$length, $data,$dataoffset

=item $retval = aio_write $fh,$offset,$length, $data,$dataoffset

=item $retval = aio_sendfile $out_fh, $in_fh, $in_offset, $length

=item $retval = aio_readahead $fh,$offset,$length

=item $status = aio_stat $fh_or_path
      
=item $status = aio_lstat $fh

=item $status = aio_unlink $pathname

=item $status = aio_rmdir $pathname

=item $entries = aio_readdir $pathname

=item ($dirs, $nondirs) = aio_scandir $path, $maxreq

=item $status = aio_fsync $fh

=item $status = aio_fdatasync $fh

=item ... = aio_xxx ...

Any additional aio requests follow the same scheme: same parameters except
you must not specify a callback but instead get the callback arguments as
return values.

=back

=head1 SEE ALSO

L<Coro::Socket> and L<Coro::Handle> for non-blocking socket operation.

=head1 AUTHOR

 Marc Lehmann <schmorp@schmorp.de>
 http://home.schmorp.de/

=cut

1
Revision:	1.13
Committed:	Mon Dec 17 06:42:05 2007 UTC (16 years, 5 months ago) by root
Branch:	MAIN
Changes since 1.12:	+7 -2 lines
Log Message:	* empty log message *
#	User	Rev	Content
1	root	1.1	=head1 NAME
2
3	root	1.2	Coro::AIO - truly asynchronous file and directrory I/O
4	root	1.1
5			=head1 SYNOPSIS
6
7			use Coro::AIO;
8
9	root	1.4	# can now use any of the aio requests your IO::AIO module supports.
10	root	1.1
11			# read 1MB of /etc/passwd, without blocking other coroutines
12			my $fh = aio_open "/etc/passwd", O_RDONLY, 0
13			or die "/etc/passwd: $!";
14			aio_read $fh, 0, 1_000_000, my $buf, 0
15			or die "aio_read: $!";
16			aio_close $fh;
17
18			=head1 DESCRIPTION
19
20			This module implements a thin wrapper around L<IO::AIO\|IO::AIO>. All of
21	root	1.2	the functions that expect a callback are being wrapped by this module.
22	root	1.1
23			The API is exactly the same as that of the corresponding IO::AIO routines,
24			except that you have to specify I<all> arguments I<except> the callback
25			argument. Instead the routines return the values normally passed to the
26	root	1.3	callback. Everything else, including C<$!> and perls stat cache, are set
27			as expected after these functions return.
28	root	1.1
29			You can mix calls to C<IO::AIO> functions with calls to this module. You
30	root	1.3	I<must not>, however, call these routines from within IO::AIO callbacks,
31			as this causes a deadlock. Start a coro inside the callback instead.
32
33			You also can, but do not need to, call C<IO::AIO::poll_cb>, as this
34			module automatically installs an event watcher for the C<IO::AIO> file
35	root	1.1	descriptor. It uses the L<AnyEvent\|AnyEvent> module for this, so please
36			refer to its documentation on how it selects an appropriate Event module.
37
38	root	1.5	All other functions exported by default by IO::AIO (e.g. C<aioreq_pri>)
39			will be exported by default by Coro::AIO, too.
40
41			Functions that can be optionally imported from IO::AIO can be imported
42			from Coro::AIO or can be called directly, e.g. C<Coro::AIO::nreqs>.
43
44	root	1.12	You cannot specify priorities with C<aioreq_pri>, as this module
45			overwrites the request priority with the current coroutine priority at all
46			times.
47
48	root	1.5	For your convienience, here are the changed function signatures for most
49			of the requests, for documentation of these functions please have a look
50			at L<IO::AIO\|the IO::AIO manual>.
51	root	1.1
52	root	1.7	The AnyEvent watcher can be disabled by executing C<undef
53			$Coro::AIO::WATCHER>. Please notify the author of when and why you think
54			this was necessary.
55
56	root	1.1	=over 4
57
58			=cut
59
60			package Coro::AIO;
61
62	root	1.8	use strict qw(subs vars);
63	root	1.2
64	root	1.1	use Coro ();
65			use AnyEvent;
66			use IO::AIO ();
67
68			use base Exporter::;
69
70	root	1.13	if (AnyEvent::detect =~ /^AnyEvent::Impl::(Coro)?EV$/) {
71			$WATCHER = EV::io IO::AIO::poll_fileno, EV::READ, \&IO::AIO::poll_cb;
72			} else {
73			our $FH; open $FH, "<&=" . IO::AIO::poll_fileno;
74			$WATCHER = AnyEvent->io (fh => $FH, poll => 'r', cb => \&IO::AIO::poll_cb);
75			}
76
77	root	1.1
78	root	1.5	our @EXPORT = @IO::AIO::EXPORT;
79			our @EXPORT_OK = @IO::AIO::EXPORT_OK;
80			our $AUTOLOAD;
81
82			{
83			my @reqs = @IO::AIO::AIO_REQ ? @IO::AIO::AIO_REQ : @EXPORT;
84			my %reqs = map +($_ => 1), @reqs;
85
86			eval
87			join "",
88			map "sub $_(" . (prototype "IO::AIO::$_") . ");",
89			grep !$reqs{$_},
90			@EXPORT, @EXPORT_OK;
91	root	1.1
92	root	1.5	for my $sub (@reqs) {
93			push @EXPORT, $sub;
94
95			my $iosub = "IO::AIO::$sub";
96			my $proto = prototype $iosub;
97	root	1.2
98	root	1.5	$proto =~ s/;?\$$// or die "$iosub: unable to remove callback slot from prototype";
99	root	1.1
100	root	1.5	eval qq{
101	root	1.3	#line 1 "Coro::AIO::$sub($proto)"
102	root	1.5	sub $sub($proto) {
103			my \$current = \$Coro::current;
104	root	1.6	my \$state;
105	root	1.5	my \@res;
106
107			push \@_, sub {
108	root	1.10	\$state = _get_state;
109	root	1.5	\@res = \@_;
110			\$current->ready;
111			};
112
113	root	1.12	aioreq_pri \$Coro::current->prio;
114	root	1.5	&$iosub;
115
116	root	1.11	&Coro::schedule;
117			&Coro::schedule while !\$state;
118	root	1.5
119	root	1.10	_set_state \$state;
120	root	1.5	wantarray ? \@res : \$res[0]
121			}
122			};
123			die if $@;
124			}
125	root	1.1	}
126
127	root	1.5	sub AUTOLOAD {
128			(my $func = $AUTOLOAD) =~ s/^.*:://;
129			*$AUTOLOAD = \&{"IO::AIO::$func"};
130			goto &$AUTOLOAD;
131			}
132	root	1.1
133			=item $fh = aio_open $pathname, $flags, $mode
134
135			=item $status = aio_close $fh
136
137			=item $retval = aio_read $fh,$offset,$length, $data,$dataoffset
138
139			=item $retval = aio_write $fh,$offset,$length, $data,$dataoffset
140
141			=item $retval = aio_sendfile $out_fh, $in_fh, $in_offset, $length
142
143			=item $retval = aio_readahead $fh,$offset,$length
144
145	root	1.2	=item $status = aio_stat $fh_or_path
146
147			=item $status = aio_lstat $fh
148
149	root	1.1	=item $status = aio_unlink $pathname
150
151			=item $status = aio_rmdir $pathname
152
153	root	1.3	=item $entries = aio_readdir $pathname
154	root	1.1
155			=item ($dirs, $nondirs) = aio_scandir $path, $maxreq
156
157			=item $status = aio_fsync $fh
158
159			=item $status = aio_fdatasync $fh
160
161	root	1.4	=item ... = aio_xxx ...
162
163			Any additional aio requests follow the same scheme: same parameters except
164			you must not specify a callback but instead get the callback arguments as
165			return values.
166
167	root	1.1	=back
168
169	root	1.2	=head1 SEE ALSO
170
171	root	1.4	L<Coro::Socket> and L<Coro::Handle> for non-blocking socket operation.
172	root	1.2
173	root	1.1	=head1 AUTHOR
174
175			Marc Lehmann <schmorp@schmorp.de>
176			http://home.schmorp.de/
177
178			=cut
179
180			1